AliveMCP

Guide · AWS SageMaker · Pipelines · ML Orchestration · MCP Tools

SageMaker Pipelines from MCP Server Tools

SageMaker Pipelines is a purpose-built ML workflow orchestrator — MCP tools can trigger pipeline executions via boto3, pass runtime parameters, and poll execution status to surface training progress to the agent. A pipeline is a DAG of steps: preprocessing (ProcessingStep), training (TrainingStep), evaluation, conditional logic (ConditionStep), model registration (RegisterModel), and deployment (LambdaStep or model deploy). Each execution is versioned and linked to the artifacts it consumed and produced — SageMaker ML Lineage automatically tracks which training dataset, code version, and hyperparameters produced each model version. For MCP servers, the main use cases are: on-demand training triggers — an MCP tool starts a pipeline with new data when a user or scheduler requests a model refresh; execution status polling — an MCP tool checks whether a running pipeline has completed and returns results; and lineage queries — an MCP tool retrieves the provenance chain for a deployed model, showing which dataset and code version it was trained on. Pipeline parameters make executions flexible: values like training_instance_type, max_epochs, or data_s3_uri can be overridden per execution without modifying the pipeline definition.

TL;DR

Call pipeline.upsert(role_arn=...) to register/update the pipeline definition. Start an execution with execution = pipeline.start(parameters={"InputDataUri": "s3://..."}). Poll status with execution.describe()["PipelineExecutionStatus"] — states are Executing, Succeeded, Failed, Stopping, Stopped. For push notification on completion, subscribe an EventBridge rule to SageMaker Pipeline Execution Status Change events.

Defining a SageMaker Pipeline for MCP-triggered training

A complete pipeline with preprocessing, training, evaluation, and conditional model registration:

import boto3
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.model_step import ModelStep
from sagemaker.processing import ScriptProcessor
from sagemaker.estimator import Estimator
from sagemaker.workflow.properties import PropertyFile

# Pipeline parameters — overridable per execution from MCP tool
input_data_uri = ParameterString(
    name="InputDataUri",
    default_value="s3://mcp-training-data/features/latest/",
)
training_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)

# Step 1: Data preprocessing
processor = ScriptProcessor(
    image_uri="763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-training:2.1.0-cpu-py310",
    command=["python3"],
    instance_type="ml.m5.large",
    instance_count=1,
    role="arn:aws:iam::123456789012:role/SageMakerPipelineRole",
)
processing_step = ProcessingStep(
    name="PreprocessMcpFeatures",
    processor=processor,
    inputs=[{"input_name": "raw_data", "source": input_data_uri}],
    outputs=[{"output_name": "processed", "destination": "s3://mcp-pipeline-artifacts/processed/"}],
    code="scripts/preprocess.py",
)

# Step 2: Model training
estimator = Estimator(
    image_uri="763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-training:2.1.0-gpu-py310",
    instance_type=training_instance_type,
    instance_count=1,
    output_path="s3://mcp-pipeline-artifacts/models/",
    role="arn:aws:iam::123456789012:role/SageMakerPipelineRole",
    hyperparameters={"epochs": max_epochs, "lr": 0.001},
)
training_step = TrainingStep(
    name="TrainMcpIntentClassifier",
    estimator=estimator,
    inputs={"training": processing_step.properties.ProcessingOutputConfig.Outputs["processed"].S3Output.S3Uri},
)

# Step 3: Evaluation — writes JSON report to S3
eval_processor = ScriptProcessor(...)
evaluation_report = PropertyFile(
    name="EvaluationReport",
    output_name="evaluation",
    path="evaluation.json",  # key in the output: {"metrics": {"accuracy": 0.91}}
)
evaluation_step = ProcessingStep(
    name="EvaluateModel",
    processor=eval_processor,
    inputs=[{"input_name": "model", "source": training_step.properties.ModelArtifacts.S3ModelArtifacts}],
    outputs=[{"output_name": "evaluation", "destination": "s3://mcp-pipeline-artifacts/eval/"}],
    property_files=[evaluation_report],
    code="scripts/evaluate.py",
)

# Step 4: Conditional registration
condition = ConditionGreaterThanOrEqualTo(
    left=JsonGet(
        step_name=evaluation_step.name,
        property_file=evaluation_report,
        json_path="metrics.accuracy",
    ),
    right=accuracy_threshold,
)
register_step = ModelStep(
    name="RegisterModel",
    step_args=model.register(
        content_types=["application/json"],
        response_types=["application/json"],
        model_package_group_name="McpIntentClassifierGroup",
        approval_status="PendingManualApproval",
    ),
)
condition_step = ConditionStep(
    name="CheckAccuracy",
    conditions=[condition],
    if_steps=[register_step],
    else_steps=[],  # fail silently if accuracy too low — upstream handles notification
)

# Assemble pipeline
pipeline = Pipeline(
    name="mcp-intent-classifier-pipeline",
    parameters=[input_data_uri, training_instance_type, max_epochs, accuracy_threshold],
    steps=[processing_step, training_step, evaluation_step, condition_step],
    sagemaker_session=sagemaker_session,
)

# Register (or update) the pipeline definition in SageMaker
pipeline.upsert(role_arn="arn:aws:iam::123456789012:role/SageMakerPipelineRole")

pipeline.upsert() creates the pipeline if it doesn't exist, or updates the definition if the pipeline already exists and the DAG has changed. SageMaker versions the definition — you can retrieve any historical version with describe_pipeline_definition_for_execution(). The ConditionStep evaluates at runtime using the output of the evaluation step — if accuracy is below threshold, the model is not registered and the pipeline succeeds without registering an artifact.

Triggering a pipeline execution from an MCP tool

The MCP tool starts the pipeline with runtime-overridden parameters and returns an execution ARN for status tracking:

import boto3
from sagemaker.workflow.pipeline import Pipeline
import sagemaker

def trigger_training_pipeline(
    input_data_uri: str,
    instance_type: str = "ml.m5.xlarge",
    max_epochs: int = 10,
    accuracy_threshold: float = 0.85,
) -> dict:
    """MCP tool: start a SageMaker training pipeline and return execution details."""
    session = sagemaker.Session(boto_session=boto3.Session(region_name="us-east-1"))
    pipeline = Pipeline(
        name="mcp-intent-classifier-pipeline",
        sagemaker_session=session,
    )

    execution = pipeline.start(
        parameters={
            "InputDataUri": input_data_uri,
            "TrainingInstanceType": instance_type,
            "MaxEpochs": max_epochs,
            "AccuracyThreshold": accuracy_threshold,
        },
        execution_description=f"Triggered by MCP tool at {datetime.utcnow().isoformat()}Z",
        parallelism_config={"MaxParallelExecutionSteps": 3},
    )

    return {
        "execution_arn": execution.arn,
        "pipeline_name": "mcp-intent-classifier-pipeline",
        "status": "Executing",
        "started_at": datetime.utcnow().isoformat() + "Z",
    }


def check_pipeline_execution(execution_arn: str) -> dict:
    """MCP tool: check pipeline execution status and return step-level details."""
    sagemaker_client = boto3.client("sagemaker", region_name="us-east-1")

    exec_detail = sagemaker_client.describe_pipeline_execution(
        PipelineExecutionArn=execution_arn
    )
    status = exec_detail["PipelineExecutionStatus"]  # Executing|Succeeded|Failed|Stopping|Stopped

    # Get step-level status for granular progress reporting
    steps = sagemaker_client.list_pipeline_execution_steps(
        PipelineExecutionArn=execution_arn
    )["PipelineExecutionSteps"]

    step_summary = [
        {
            "name": s["StepName"],
            "status": s["StepStatus"],  # Starting|Executing|Stopped|Failed|Succeeded|Skipped
            "start_time": s.get("StartTime", ""),
            "end_time": s.get("EndTime", ""),
        }
        for s in steps
    ]

    result = {"status": status, "steps": step_summary}

    if status == "Failed":
        result["failure_reason"] = exec_detail.get("FailureReason", "unknown")

    if status == "Succeeded":
        # Retrieve evaluation metrics from the pipeline execution properties
        for step in steps:
            if step["StepName"] == "EvaluateModel" and step["StepStatus"] == "Succeeded":
                result["model_accuracy"] = step.get(
                    "Metadata", {}
                ).get("ProcessingJob", {}).get("Arn", "")
                break

    return result

The execution_description parameter is surfaced in the SageMaker console and appears in CloudWatch Logs — use it to embed the MCP session ID or user context so you can trace which agent invocation triggered a given training run.

EventBridge notifications for pipeline completion

For MCP tools that submit a pipeline and return immediately, use EventBridge to push a notification when the execution completes rather than polling:

# EventBridge rule — matches SageMaker Pipeline execution status changes
# Create via CDK or CloudFormation:
{
  "source": ["aws.sagemaker"],
  "detail-type": ["SageMaker Model Building Pipeline Execution Status Change"],
  "detail": {
    "pipelineArn": ["arn:aws:sagemaker:us-east-1:123456789012:pipeline/mcp-intent-classifier-pipeline"],
    "currentPipelineExecutionStatus": ["Succeeded", "Failed", "Stopped"]
  }
}

# EventBridge event payload structure:
# {
#   "version": "0",
#   "source": "aws.sagemaker",
#   "detail-type": "SageMaker Model Building Pipeline Execution Status Change",
#   "detail": {
#     "pipelineArn": "arn:aws:sagemaker:...:pipeline/mcp-intent-classifier-pipeline",
#     "pipelineExecutionArn": "arn:aws:sagemaker:...:pipeline/.../execution/...",
#     "currentPipelineExecutionStatus": "Succeeded",
#     "previousPipelineExecutionStatus": "Executing",
#     "executionDescription": "Triggered by MCP tool at 2026-09-26T10:00:00Z",
#     "pipelineExecutionDisplayName": "...",
#   }
# }

# Lambda handler for the EventBridge target
def handle_pipeline_completion(event, context):
    detail = event["detail"]
    execution_arn = detail["pipelineExecutionArn"]
    status = detail["currentPipelineExecutionStatus"]
    description = detail.get("executionDescription", "")

    # Extract MCP session ID from execution description
    # e.g., "session_id=abc123; Triggered by MCP tool at ..."
    session_id = None
    for part in description.split(";"):
        if "session_id=" in part:
            session_id = part.split("=", 1)[1].strip()

    # Notify the waiting MCP session via SQS, DynamoDB, or WebSocket
    if session_id:
        sqs = boto3.client("sqs")
        sqs.send_message(
            QueueUrl="https://sqs.us-east-1.amazonaws.com/123456789012/mcp-pipeline-results",
            MessageBody=json.dumps({
                "session_id": session_id,
                "execution_arn": execution_arn,
                "status": status,
            }),
        )

Embedding a correlation ID (session ID, request ID) in the execution_description allows the EventBridge Lambda to route completion notifications back to the right MCP tool session. This is the recommended pattern for long-running training jobs — the MCP tool returns to the agent immediately after triggering, and a separate check_pipeline_status tool polls DynamoDB or SQS for the result.

Pipeline lineage queries from MCP tools

SageMaker ML Lineage automatically tracks the artifacts consumed and produced by each step. MCP tools can query lineage to answer provenance questions:

def get_model_lineage(model_package_arn: str) -> dict:
    """
    Return the training provenance for a registered model version.
    Answers: which dataset, code version, and pipeline execution produced this model?
    """
    sagemaker_client = boto3.client("sagemaker")

    # List associations upstream of the model package
    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"]  # Artifact, Execution, 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", ""),
                "started_at": str(exec_detail.get("CreationTime", "")),
                "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", ""),
                "properties": artifact.get("Properties", {}),
            })

    return lineage

ML Lineage uses a graph model — artifacts (datasets, models) and executions (training jobs, processing jobs) are nodes; associations are directed edges. The ContributedTo association type links an upstream artifact or execution to a downstream model. Use list_associations(SourceArn=execution_arn) to traverse forward (what artifacts did this execution produce) or list_associations(DestinationArn=model_arn) to traverse backward (what went into producing this model).

Monitor MCP endpoints that serve pipeline-trained models

A SageMaker Pipeline can register a new model version automatically after training. If a newly deployed model version causes the MCP endpoint to return errors or time out, AliveMCP catches the regression immediately — alerting your team before users report failures from the freshly trained model.

Join the waitlist →