AWS SageMaker for MCP Server Tools: Inference Deployment, ML Context, and Training Lifecycle

Pattern 1 — Inference deployment choices

An MCP server tool that calls a ML model has three execution path choices in SageMaker, and the right choice depends on the latency budget, traffic pattern, and whether you need private model hosting. Getting this decision wrong costs either latency (wrong mode) or money (wrong instance configuration).

Real-time inference — synchronous calls under 60 seconds

The real-time endpoint is the right choice for conversational MCP tools: intent classification, entity extraction, short-context generation, or any model call where the agent needs the result within the same tool execution turn. The call is synchronous — the MCP tool handler blocks until the model returns or an exception fires:

import boto3
import json

# Use the runtime client, not boto3.client("sagemaker")
# The control plane client is for endpoint management, not inference
sagemaker_runtime = boto3.client("sagemaker-runtime", region_name="us-east-1")

def classify_intent(user_input: str) -> dict:
    payload = {
        "inputs": user_input,
        "parameters": {"max_length": 128, "return_all_scores": True},
    }
    try:
        response = sagemaker_runtime.invoke_endpoint(
            EndpointName="mcp-intent-classifier-prod",
            ContentType="application/json",
            Accept="application/json",
            Body=json.dumps(payload),
        )
        # StreamingBody must be .read() — not consuming it causes connection
        # pool exhaustion in long-running MCP server processes
        result = json.loads(response["Body"].read())
        return {"intent": result[0]["label"], "score": result[0]["score"]}

    except sagemaker_runtime.exceptions.ModelError as e:
        # Container returned 4xx or 5xx — check OriginalStatusCode and OriginalMessage
        # Common causes: malformed input shape, wrong ContentType, max_sequence_length exceeded
        raise RuntimeError(
            f"Model error {e.response.get('OriginalStatusCode')}: "
            f"{e.response.get('OriginalMessage')}"
        ) from e
    except sagemaker_runtime.exceptions.InternalFailure as e:
        # SageMaker infrastructure — retry with backoff, the model itself may be healthy
        raise RuntimeError(f"Infrastructure error: {e}") from e
    except sagemaker_runtime.exceptions.ServiceUnavailable as e:
        # Endpoint initializing or overloaded — back off, do not mark as permanent failure
        raise RuntimeError(f"Endpoint unavailable: {e}") from e

The three exception classes are not interchangeable. ModelError means the container processed the request and returned a non-2xx status — retrying without changing the input will reproduce the same error. InternalFailure and ServiceUnavailable are transient infrastructure conditions where the same request should succeed after a wait.

ContentType selection is the other failure source: the header must match what the model container's /invocations handler expects. TGI (Text Generation Inference) containers accept application/json with an inputs key. Built-in XGBoost and Linear Learner models accept text/csv with a raw feature vector. PyTorch TorchServe containers may require application/x-pytorch with a serialized tensor. A mismatch produces a ModelError with status 415.

Auto-scaling real-time endpoints for MCP traffic bursts

SageMaker endpoints have no automatic scaling — you configure Application Auto Scaling separately. The standard target metric is SageMakerVariantInvocationsPerInstance:

autoscaling = boto3.client("application-autoscaling", region_name="us-east-1")
resource_id = f"endpoint/{ENDPOINT_NAME}/variant/AllTraffic"

autoscaling.register_scalable_target(
    ServiceNamespace="sagemaker",
    ResourceId=resource_id,
    ScalableDimension="sagemaker:variant:DesiredInstanceCount",
    MinCapacity=1,
    MaxCapacity=8,
)
autoscaling.put_scaling_policy(
    PolicyName="mcp-invocations-policy",
    ServiceNamespace="sagemaker",
    ResourceId=resource_id,
    ScalableDimension="sagemaker:variant:DesiredInstanceCount",
    PolicyType="TargetTrackingScaling",
    TargetTrackingScalingPolicyConfiguration={
        "TargetValue": 70.0,
        "PredefinedMetricSpecification": {
            "PredefinedMetricType": "SageMakerVariantInvocationsPerInstance",
        },
        "ScaleInCooldown": 300,   # 5 minutes — avoid instance churn
        "ScaleOutCooldown": 60,   # 1 minute — scale out fast for MCP bursts
    },
)

The asymmetric cooldown is intentional: new instances take 2–5 minutes to load the model container. Scaling in too aggressively means the next burst waits for a cold start. MinCapacity: 1 keeps one warm instance always available; setting it to 0 trades cost savings for a cold-start penalty on the first MCP call after an idle period.

Async inference — model calls from 30 seconds to 1 hour

Async inference decouples submission from result — the MCP tool uploads input to S3, calls invoke_endpoint_async, and receives an InferenceId and OutputLocation immediately. SageMaker queues the request and processes it when a compute instance is available:

import uuid

s3 = boto3.client("s3")
sagemaker_runtime = boto3.client("sagemaker-runtime", region_name="us-east-1")

def submit_document_for_processing(document_text: str) -> dict:
    job_id = str(uuid.uuid4())
    s3.put_object(
        Bucket="mcp-async-io",
        Key=f"inputs/{job_id}.json",
        Body=json.dumps({"document": document_text}),
        ContentType="application/json",
    )
    response = sagemaker_runtime.invoke_endpoint_async(
        EndpointName="mcp-doc-processor-async",
        ContentType="application/json",
        InputLocation=f"s3://mcp-async-io/inputs/{job_id}.json",
        InvocationTimeoutSecs=300,   # fail after 5 minutes if no model response
        RequestTTLSecs=600,          # drop from queue if not dequeued within 10 minutes
                                     # set ≥ 2× InvocationTimeoutSecs to avoid race
    )
    return {
        "inference_id": response["InferenceId"],
        "output_location": response["OutputLocation"],
    }

RequestTTLSecs is the time a request can wait in the async queue before being discarded. If RequestTTLSecs is less than InvocationTimeoutSecs, a request that starts near the TTL deadline can be dropped mid-execution — always set RequestTTLSecs to at least twice InvocationTimeoutSecs.

For the simplest integration, poll S3 for the output file:

import time

def poll_async_result(output_location: str, max_wait: float = 300.0) -> dict:
    # Parse s3://bucket/key
    parts = output_location[5:].split("/", 1)
    bucket, key = parts[0], parts[1]
    base_key = key.rsplit(".", 1)[0] if "." in key else key
    failure_key = f"{base_key}.failure"

    s3_client = boto3.client("s3")
    deadline = time.monotonic() + max_wait
    interval = 2.0
    while time.monotonic() < deadline:
        try:
            obj = s3_client.get_object(Bucket=bucket, Key=key)
            return json.loads(obj["Body"].read())
        except s3_client.exceptions.NoSuchKey:
            pass
        try:
            fail = s3_client.get_object(Bucket=bucket, Key=failure_key)
            raise RuntimeError(f"Inference failed: {fail['Body'].read().decode()}")
        except s3_client.exceptions.NoSuchKey:
            pass
        time.sleep(interval)
        interval = min(interval * 1.2, 10.0)  # gentle backoff, cap at 10s
    raise TimeoutError(f"No result after {max_wait}s at {output_location}")

For jobs over 60 seconds, the better pattern is to return the InferenceId as part of the tool response and let the LLM call a check_job_status MCP tool in a follow-up turn — this avoids holding an open connection for minutes. Wire an SNS notification to a Lambda that stores results in DynamoDB, and have check_job_status query DynamoDB by InferenceId.

JumpStart — private foundation model hosting

JumpStart deploys a foundation model from the SageMaker registry into your own AWS account in one call. No per-token pricing, no data leaving your VPC, and no usage logs sent to a third-party API. The trade-off: idle instance cost even during quiet periods, and model availability is 2–4 weeks behind public releases.

from sagemaker.jumpstart.model import JumpStartModel
import sagemaker, boto3

session = sagemaker.Session(boto_session=boto3.Session(region_name="us-east-1"))
model = JumpStartModel(
    model_id="meta-textgeneration-llama-3-8b-instruct",
    model_version="*",     # pin a specific version in production
    role="arn:aws:iam::123456789012:role/SageMakerJumpStartRole",
    sagemaker_session=session,
)
predictor = model.deploy(
    initial_instance_count=1,
    instance_type="ml.g5.12xlarge",   # 4× A10G, sufficient for Llama 3 8B in fp16
    endpoint_name="mcp-llama3-8b",
)

Minimum instance requirements by model family (2026): Llama 3 8B Instruct → ml.g5.2xlarge (4-bit) or ml.g5.12xlarge (fp16); Llama 3 70B Instruct → ml.g5.48xlarge (4-bit); Mistral 7B Instruct → ml.g5.2xlarge; Falcon 40B → ml.g5.12xlarge; Stable Diffusion 2.1 → ml.g5.2xlarge. Deploying on an undersized instance causes the container to crash with an InternalFailure that is not obviously attributable to instance size in the error message.

Payload format differs by model family — the MCP tool handler must adapt:

sagemaker_runtime = boto3.client("sagemaker-runtime", region_name="us-east-1")

# Llama 3 (TGI container) — messages API format
def call_llama3(user_prompt: str, system_prompt: str = "") -> str:
    messages = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": user_prompt})
    payload = {
        "inputs": messages,
        "parameters": {
            "max_new_tokens": 512,
            "temperature": 0.7,
            "return_full_text": False,   # omit this → TGI echoes the full input + output
        },
    }
    response = sagemaker_runtime.invoke_endpoint(
        EndpointName="mcp-llama3-8b",
        ContentType="application/json",
        Body=json.dumps(payload),
    )
    return json.loads(response["Body"].read())[0]["generated_text"]

# Mistral (different prompt template — [INST] markers, NOT the messages API)
def call_mistral(user_prompt: str) -> str:
    prompt = f"[INST] {user_prompt} [/INST]"
    payload = {"inputs": prompt, "parameters": {"max_new_tokens": 512}}
    response = sagemaker_runtime.invoke_endpoint(
        EndpointName="mcp-mistral-7b",
        ContentType="application/json",
        Body=json.dumps(payload),
    )
    return json.loads(response["Body"].read())[0]["generated_text"]

Inference components — multi-model GPU sharing

When an MCP server needs to call several foundation models, Inference Components allow multiple models to share one large GPU instance rather than paying for a dedicated endpoint per model:

sagemaker_client = boto3.client("sagemaker")

# Create a shared endpoint (no model reference in ProductionVariants)
sagemaker_client.create_endpoint_config(
    EndpointConfigName="mcp-shared-gpu-config",
    ProductionVariants=[{
        "VariantName": "AllTraffic",
        "InstanceType": "ml.p4d.24xlarge",   # 8× A100 GPUs total
        "InitialInstanceCount": 1,
        "RoutingConfig": {"RoutingStrategy": "LEAST_OUTSTANDING_REQUESTS"},
    }],
    ExecutionRoleArn="arn:aws:iam::123456789012:role/SageMakerJumpStartRole",
)

# Allocate GPU slices per model via inference components
sagemaker_client.create_inference_component(
    InferenceComponentName="mcp-llama3-70b",
    EndpointName="mcp-shared-gpu-endpoint",
    VariantName="AllTraffic",
    Specification={
        "ModelName": "meta-textgeneration-llama-3-70b-instruct",
        "ComputeResourceRequirements": {
            "NumberOfAcceleratorDevicesRequired": 4,   # half the A100s
            "MinMemoryRequiredInMb": 180000,
        },
    },
    RuntimeConfig={"CopyCount": 1},
)

# Invoke a specific component by name
response = sagemaker_runtime.invoke_endpoint(
    EndpointName="mcp-shared-gpu-endpoint",
    InferenceComponentName="mcp-llama3-70b",   # route to this specific model
    ContentType="application/json",
    Body=json.dumps({"inputs": messages, "parameters": {"max_new_tokens": 256}}),
)

The economics: four foundation models on a single ml.p4d.24xlarge cost approximately $30/hour; four dedicated ml.g5.48xlarge instances cost approximately $50/hour. Inference components are the right pattern whenever an MCP server exposes multiple AI-powered tools backed by different models.

Pattern 2 — ML context management with Feature Store

Feature Store solves the training/serving skew problem — the same feature definitions and feature values are used during both model training and MCP tool inference. Without it, the feature transformation logic lives in two places (training pipeline and MCP tool handler) and diverges over time, silently degrading model accuracy in production.

Online store reads at inference time

The online store is a DynamoDB-backed key-value layer optimized for single-digit millisecond lookups. Use the sagemaker-featurestore-runtime client — this is a separate boto3 service name from the SageMaker control plane:

import boto3
import json

featurestore_runtime = boto3.client(
    "sagemaker-featurestore-runtime",
    region_name="us-east-1",
)

def get_user_features(user_id: str) -> dict:
    """Retrieve precomputed features for inference. Returns cold-start defaults on miss."""
    try:
        response = featurestore_runtime.get_record(
            FeatureGroupName="mcp-user-behavior-features",
            RecordIdentifierValueAsString=user_id,
            FeatureNames=[           # request only what the model needs
                "tool_calls_last_7d",
                "avg_session_duration_s",
                "error_rate_last_24h",
                "intent_embedding_json",
            ],
        )
        # All values are returned as strings regardless of declared FeatureType
        # Must cast to the correct Python type before passing to the model
        raw = {f["FeatureName"]: f["ValueAsString"] for f in response["Record"]}
        return {
            "tool_calls_last_7d": int(raw["tool_calls_last_7d"]),
            "avg_session_duration_s": float(raw["avg_session_duration_s"]),
            "error_rate_last_24h": float(raw["error_rate_last_24h"]),
            "intent_embedding": json.loads(raw["intent_embedding_json"]),
        }
    except featurestore_runtime.exceptions.ResourceNotFound:
        # Entity has no features yet — provide cold-start defaults
        return {
            "tool_calls_last_7d": 0,
            "avg_session_duration_s": 0.0,
            "error_rate_last_24h": 0.0,
            "intent_embedding": [],
        }

ResourceNotFound for a missing entity is not an error in MCP tool context — new users, new sessions, and new documents all arrive before any feature has been written. Always handle it with sensible defaults rather than surfacing it as a tool failure.

BatchGetRecord for multi-entity tool calls

When an MCP tool needs features for multiple entities in one call — all users in a session, all documents in a batch, all tools in a catalog — use batch_get_record to retrieve up to 100 records across up to 10 feature groups in a single API call:

def get_batch_features(user_ids: list[str], tool_ids: list[str]) -> dict:
    response = featurestore_runtime.batch_get_record(
        Identifiers=[
            {
                "FeatureGroupName": "mcp-user-behavior-features",
                "RecordIdentifiersValueAsString": user_ids,
                "FeatureNames": ["tool_calls_last_7d", "error_rate_last_24h"],
            },
            {
                "FeatureGroupName": "mcp-tool-popularity-features",
                "RecordIdentifiersValueAsString": tool_ids,
                "FeatureNames": ["invocations_last_7d", "avg_latency_ms", "success_rate"],
            },
        ]
    )

    result = {"users": {}, "tools": {}}
    for batch in response["Records"]:
        fg = batch["FeatureGroupName"]
        entity_id = batch["RecordIdentifier"]
        features = {f["FeatureName"]: f["ValueAsString"] for f in batch["Record"]}
        if "user" in fg:
            result["users"][entity_id] = features
        else:
            result["tools"][entity_id] = features

    # Errors list contains entities with no record — NOT raised as an exception
    # Failing to check this means silently missing features for new entities
    for error in response.get("Errors", []):
        entity_id = error["RecordIdentifier"]
        # fill defaults based on feature group
        pass

    return result

The Errors list in the response is the most common oversight: missing entities appear here rather than raising an exception, so if you don't check response["Errors"], your MCP tool silently processes some entities without features — producing degraded model outputs without any error signal.

PutRecord — writing features back from MCP tools

MCP tools that take actions can update feature values after execution, keeping the online store fresh for subsequent inference calls:

from datetime import datetime, timezone

def record_tool_execution(user_id: str, tool_category: str, success: bool) -> None:
    event_time = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    featurestore_runtime.put_record(
        FeatureGroupName="mcp-user-behavior-features",
        Record=[
            {"FeatureName": "user_id",                 "ValueAsString": user_id},
            {"FeatureName": "event_time",              "ValueAsString": event_time},
            {"FeatureName": "preferred_tool_category", "ValueAsString": tool_category},
            # Partial updates are supported — omit features you're not changing
            # Online store merges; offline store appends a new row
        ],
        TargetStores=["OnlineStore"],  # write only online for lowest latency
    )

EventTimeFeatureName must be ISO 8601 in UTC — SageMaker rejects other formats with a validation error. The online store merges the new values with the existing record for that RecordIdentifierValueAsString. The offline store always appends a new row, which is why every Athena query over the offline store needs AND NOT is_deleted: Feature Store implements DeleteRecord as a soft delete that appends a row with is_deleted = true.

Offline store queries for analytics MCP tools

MCP tools that serve analytics data or generate training exports can query the offline store via Athena. SageMaker automatically creates a Glue catalog table when the feature group is created with OfflineStoreConfig:

athena = boto3.client("athena", region_name="us-east-1")

def query_user_feature_history(user_id: str, since_date: str) -> list[dict]:
    query = f"""
    SELECT user_id, event_time, tool_calls_last_7d, error_rate_last_24h, write_time
    FROM "sagemaker_featurestore"."mcp_user_behavior_features"
    WHERE user_id = '{user_id}'
      AND event_time >= '{since_date}'
      AND NOT is_deleted   -- required: Feature Store soft-deletes via is_deleted flag
    ORDER BY event_time ASC
    """
    execution = athena.start_query_execution(
        QueryString=query,
        QueryExecutionContext={"Database": "sagemaker_featurestore"},
        ResultConfiguration={"OutputLocation": "s3://mcp-athena-results/"},
    )
    query_id = execution["QueryExecutionId"]
    # Poll until complete, then paginate results via get_query_results
    ...

The Glue table name is the feature group name with hyphens replaced by underscores (e.g., mcp-user-behavior-features → mcp_user_behavior_features). The TableFormat: "Iceberg" option enables time-travel queries — you can query feature values as of any past timestamp, which is the correct approach for point-in-time feature extraction when building training datasets.

Pattern 3 — ML lifecycle from MCP tools

MCP tools can be the control plane for the model training loop — triggering pipelines, monitoring executions, and querying provenance. SageMaker Pipelines provides a versioned, auditable DAG that supports runtime parameter overrides, which is precisely what an agent needs when deciding to refresh a model on demand.

Defining a pipeline for MCP-triggered training

A complete pipeline with preprocessing, training, evaluation, and conditional model registration. The ConditionStep gates registration on evaluation accuracy — if the model doesn't meet the threshold, it is not registered and no downstream actions occur:

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.parameters import ParameterString, ParameterFloat, ParameterInteger
from sagemaker.workflow.properties import PropertyFile
from sagemaker.workflow.functions import JsonGet

# Parameters overridable per execution from the MCP tool
input_data_uri     = ParameterString(name="InputDataUri", default_value="s3://...")
instance_type      = ParameterString(name="TrainingInstanceType", default_value="ml.m5.xlarge")
max_epochs         = ParameterInteger(name="MaxEpochs", default_value=10)
accuracy_threshold = ParameterFloat(name="AccuracyThreshold", default_value=0.85)

# ... define ProcessingStep, TrainingStep, evaluation ProcessingStep ...

# Evaluation step writes {"metrics": {"accuracy": 0.91}} to S3 as a PropertyFile
evaluation_report = PropertyFile(
    name="EvaluationReport",
    output_name="evaluation",
    path="evaluation.json",
)

# ConditionStep evaluates accuracy at runtime without custom Lambda glue
condition = ConditionGreaterThanOrEqualTo(
    left=JsonGet(
        step_name=evaluation_step.name,
        property_file=evaluation_report,
        json_path="metrics.accuracy",
    ),
    right=accuracy_threshold,
)
condition_step = ConditionStep(
    name="CheckAccuracy",
    conditions=[condition],
    if_steps=[register_step],  # only registers if accuracy >= threshold
    else_steps=[],
)

pipeline = Pipeline(
    name="mcp-intent-classifier-pipeline",
    parameters=[input_data_uri, instance_type, max_epochs, accuracy_threshold],
    steps=[processing_step, training_step, evaluation_step, condition_step],
    sagemaker_session=sagemaker_session,
)
pipeline.upsert(role_arn="arn:aws:iam::123456789012:role/SageMakerPipelineRole")

pipeline.upsert() creates the pipeline on first call and updates the definition on subsequent calls. SageMaker versions the definition automatically — each execution records which version of the DAG was used.

Triggering and polling from an MCP tool

The MCP tool starts the pipeline with runtime parameters and embeds a correlation ID in execution_description — this is how the EventBridge callback routes the completion notification back to the originating MCP session:

import sagemaker
from sagemaker.workflow.pipeline import Pipeline
from datetime import datetime, timezone

def trigger_training(input_s3_uri: str, session_id: str) -> dict:
    """MCP tool: start a training pipeline and return immediately."""
    pipeline = Pipeline(
        name="mcp-intent-classifier-pipeline",
        sagemaker_session=sagemaker.Session(),
    )
    execution = pipeline.start(
        parameters={
            "InputDataUri": input_s3_uri,
            "MaxEpochs": 15,
            "AccuracyThreshold": 0.88,
        },
        # Embed session_id here — EventBridge Lambda reads it to route notifications
        execution_description=f"session_id={session_id}; triggered at {datetime.now(timezone.utc).isoformat()}",
        parallelism_config={"MaxParallelExecutionSteps": 3},
    )
    return {"execution_arn": execution.arn, "status": "Executing"}

def check_training_status(execution_arn: str) -> dict:
    """MCP tool: return step-level progress for a running pipeline."""
    sagemaker_client = boto3.client("sagemaker")
    detail = sagemaker_client.describe_pipeline_execution(
        PipelineExecutionArn=execution_arn
    )
    steps = sagemaker_client.list_pipeline_execution_steps(
        PipelineExecutionArn=execution_arn
    )["PipelineExecutionSteps"]
    return {
        "status": detail["PipelineExecutionStatus"],
        "failure_reason": detail.get("FailureReason"),
        "steps": [
            {"name": s["StepName"], "status": s["StepStatus"]}
            for s in steps
        ],
    }

PipelineExecutionStatus cycles through: Executing → Succeeded / Failed / Stopped. Step-level status from list_pipeline_execution_steps provides granularity: each step shows Starting, Executing, Succeeded, Failed, or Skipped (when a ConditionStep branch is not taken).

EventBridge for push-notification on pipeline completion

For long-running training jobs, polling from the MCP tool is wasteful. Use an EventBridge rule that matches SageMaker Pipeline status change events and routes the completion notification back to the originating session via SQS:

# EventBridge rule (CloudFormation / CDK):
# {
#   "source": ["aws.sagemaker"],
#   "detail-type": ["SageMaker Model Building Pipeline Execution Status Change"],
#   "detail": {
#     "currentPipelineExecutionStatus": ["Succeeded", "Failed", "Stopped"]
#   }
# }

# Lambda target for the EventBridge rule
def handle_pipeline_completion(event, context):
    detail = event["detail"]
    description = detail.get("executionDescription", "")

    # Extract the session_id embedded in execution_description at trigger time
    session_id = None
    for part in description.split(";"):
        part = part.strip()
        if part.startswith("session_id="):
            session_id = part.split("=", 1)[1].strip()

    if session_id:
        sqs = boto3.client("sqs")
        sqs.send_message(
            QueueUrl="https://sqs.us-east-1.amazonaws.com/.../mcp-pipeline-results",
            MessageBody=json.dumps({
                "session_id": session_id,
                "execution_arn": detail["pipelineExecutionArn"],
                "status": detail["currentPipelineExecutionStatus"],
            }),
        )

# MCP tool: check_training_result (polls SQS or DynamoDB instead of SageMaker)
def check_training_result(session_id: str) -> dict:
    dynamodb = boto3.resource("dynamodb")
    table = dynamodb.Table("mcp-pipeline-results")
    item = table.get_item(Key={"session_id": session_id}).get("Item")
    if item is None:
        return {"status": "pending"}
    return {"status": item["pipeline_status"], "execution_arn": item["execution_arn"]}

The executionDescription field in the EventBridge event payload is exactly what was passed to execution.start(execution_description=...). Use it as the correlation channel between the submitting MCP tool and the completion callback — without it, the Lambda has no way to route the result back to the session that triggered the pipeline.

ML Lineage queries from MCP tools

SageMaker ML Lineage automatically records the associations between artifacts (datasets, models), executions (training jobs, processing jobs), and model package registrations. An MCP tool can traverse this graph to answer provenance questions without any custom tracking:

def get_model_provenance(model_package_arn: str) -> dict:
    """Return the training history for a registered model version."""
    sagemaker_client = boto3.client("sagemaker")

    # Traverse backward: what artifacts and executions contributed to this model?
    associations = sagemaker_client.list_associations(
        DestinationArn=model_package_arn,
        AssociationType="ContributedTo",
    )["AssociationSummaries"]

    lineage = {"model_package_arn": model_package_arn, "ancestors": []}
    for assoc in associations:
        source_arn = assoc["SourceArn"]
        source_type = assoc["SourceType"]   # Execution | Artifact | Context | Action

        if source_type == "Execution":
            exec_detail = sagemaker_client.describe_pipeline_execution(
                PipelineExecutionArn=source_arn
            )
            lineage["ancestors"].append({
                "type": "pipeline_execution",
                "arn": source_arn,
                "pipeline": exec_detail.get("PipelineArn", ""),
                "status": exec_detail.get("PipelineExecutionStatus", ""),
                "description": exec_detail.get("PipelineExecutionDescription", ""),
            })
        elif source_type == "Artifact":
            artifact = sagemaker_client.describe_artifact(ArtifactArn=source_arn)
            lineage["ancestors"].append({
                "type": "artifact",
                "artifact_type": artifact.get("ArtifactType", ""),
                "source_uri": artifact.get("Source", {}).get("SourceUri", ""),
            })
    return lineage

The graph traversal direction matters: list_associations(DestinationArn=model_arn) traverses backward (what went into this model); list_associations(SourceArn=execution_arn) traverses forward (what did this execution produce). The ContributedTo association type links upstream artifacts and executions to the model they produced.

Consolidated failure modes across the SageMaker arc

Production checklists

Real-time inference checklist

• response["Body"].read() is called in every code path, including exception handlers
• ModelError, InternalFailure, and ServiceUnavailable have separate handling — retries only on InternalFailure and ServiceUnavailable
• Application Auto Scaling configured with asymmetric cooldowns (ScaleOutCooldown: 60, ScaleInCooldown: 300)
• MinCapacity set based on latency tolerance — not 0 unless scale-to-zero is explicitly accepted
• MME pre-warm called before the first user interaction for latency-sensitive TargetModels
• InvokedProductionVariant logged for all variant-split endpoints to enable offline evaluation

Async inference checklist

• RequestTTLSecs ≥ 2 × InvocationTimeoutSecs
• Both success key and .failure key checked in every poll iteration
• Long jobs (>60s) return InferenceId as a tool result and use a separate check_job_status tool rather than blocking the MCP connection
• SNS NotificationConfig configured for both SuccessTopic and ErrorTopic in production
• MaxConcurrentInvocationsPerInstance set appropriately: 1–2 for GPU LLMs, 4–8 for CPU document processing

Feature Store checklist

• ResourceNotFound handled with cold-start defaults in every get_record call
• response["Errors"] checked and defaults filled in every batch_get_record call
• All feature values cast from string to correct Python type immediately after retrieval
• EventTimeFeatureName written as ISO 8601 UTC (e.g., 2026-09-26T10:00:00.000000Z)
• Every offline store Athena query includes AND NOT is_deleted
• TableFormat: "Iceberg" set on feature groups that require time-travel training data generation

Pipelines and Lineage checklist

• Correlation ID embedded in execution_description on every pipeline.start() call
• EventBridge rule targets a Lambda that extracts the correlation ID and routes to SQS/DynamoDB
• Evaluation PropertyFile path and JsonGet json_path validated against the actual evaluation script output before deployment
• ConditionStep else_steps includes a notification step if below-threshold accuracy should trigger an alert
• ML Lineage list_associations traversal direction understood: DestinationArn for backward (what built this model), SourceArn for forward (what did this execution produce)
• Pipeline definition pinned by version in production — pipeline.upsert() only called from CI/CD, not from MCP tool handlers at runtime