Guide · AWS SageMaker · Feature Store · MCP Tools

SageMaker Feature Store in MCP Server Tools

SageMaker Feature Store provides a centralized repository for ML features with two storage tiers: the online store for single-digit millisecond reads at inference time, and the offline store (S3 + Glue catalog) for batch training data retrieval. For MCP tools that run ML inference, Feature Store solves the training/serving skew problem — the same feature definitions, transformation logic, and feature values are used during both model training and online inference. An MCP tool handler calls GetRecord to retrieve the precomputed feature vector for a given entity (user ID, document ID, session ID) and passes it directly to the model endpoint. The tool can also call PutRecord to write new feature observations back to the store — for example, updating a user's behavioral features after the MCP tool executes an action. Three Feature Store patterns cover most MCP server use cases: read-time feature retrieval for passing context to model calls; batch feature reads with BatchGetRecord for tools that score multiple entities in one call; and offline store queries via Athena for MCP tools that serve analytics results or training data exports.

TL;DR

Use boto3.client("sagemaker-featurestore-runtime").get_record(FeatureGroupName=..., RecordIdentifierValueAsString=entity_id) for online reads. Use batch_get_record for multiple entities in one API call. Write features with put_record(FeatureGroupName=..., Record=[{FeatureName:..., ValueAsString:...}]). Query the offline store via Athena using the Glue catalog table that SageMaker creates automatically when EnableOnlineStore is set with OfflineStoreConfig.

Feature group schema and creation

A feature group defines the schema: each feature has a name and type (String, Fractional, Integral). Every record requires a RecordIdentifierFeatureName (entity key) and an EventTimeFeatureName (ISO 8601 timestamp for the observation):

import boto3

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

# Create a feature group for user behavioral features used in MCP recommendation tools
sagemaker.create_feature_group(
    FeatureGroupName="mcp-user-behavior-features",
    RecordIdentifierFeatureName="user_id",
    EventTimeFeatureName="event_time",
    FeatureDefinitions=[
        {"FeatureName": "user_id",                 "FeatureType": "String"},
        {"FeatureName": "event_time",              "FeatureType": "String"},
        # Behavioral signals
        {"FeatureName": "tool_calls_last_7d",      "FeatureType": "Integral"},
        {"FeatureName": "avg_session_duration_s",  "FeatureType": "Fractional"},
        {"FeatureName": "preferred_tool_category", "FeatureType": "String"},
        {"FeatureName": "error_rate_last_24h",     "FeatureType": "Fractional"},
        {"FeatureName": "api_quota_used_pct",      "FeatureType": "Fractional"},
        # Precomputed embedding (stored as JSON string)
        {"FeatureName": "intent_embedding_json",   "FeatureType": "String"},
    ],
    OnlineStoreConfig={
        "EnableOnlineStore": True,
        "SecurityConfig": {
            "KmsKeyId": "arn:aws:kms:us-east-1:123456789012:key/...",
        },
    },
    OfflineStoreConfig={
        "S3StorageConfig": {
            "S3Uri": "s3://mcp-feature-store/offline/",
            "KmsKeyId": "arn:aws:kms:us-east-1:123456789012:key/...",
        },
        "DisableGlueTableCreation": False,  # auto-create Glue table for Athena
        "TableFormat": "Iceberg",  # Iceberg enables time-travel queries
    },
    RoleArn="arn:aws:iam::123456789012:role/SageMakerFeatureStoreRole",
    Description="User behavioral features for MCP tool personalization",
    Tags=[{"Key": "service", "Value": "alivemcp"}, {"Key": "env", "Value": "prod"}],
)

TableFormat: "Iceberg" enables time-travel queries in the offline store — you can query feature values as of any past timestamp, which is valuable for point-in-time correct training data generation. Without Iceberg, the offline store uses Parquet files with Hive partitioning.

Online store reads from MCP tool handlers

The online store is a DynamoDB-backed key-value store optimized for single-record lookups. Access it via the sagemaker-featurestore-runtime client (different service name from the control plane client):

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 user features for an MCP tool inference call."""
    try:
        response = featurestore_runtime.get_record(
            FeatureGroupName="mcp-user-behavior-features",
            RecordIdentifierValueAsString=user_id,
            # Optional: request only specific features to reduce payload
            FeatureNames=[
                "tool_calls_last_7d",
                "avg_session_duration_s",
                "error_rate_last_24h",
                "intent_embedding_json",
            ],
        )
        # Convert list of {FeatureName, ValueAsString} to a flat dict
        return {
            feat["FeatureName"]: feat["ValueAsString"]
            for feat in response["Record"]
        }
    except featurestore_runtime.exceptions.ResourceNotFound:
        # Entity has no features yet — return cold-start defaults
        return {
            "tool_calls_last_7d": "0",
            "avg_session_duration_s": "0.0",
            "error_rate_last_24h": "0.0",
            "intent_embedding_json": "[]",
        }

# Combine features with real-time context before calling the model
def recommend_tools(user_id: str, current_query: str) -> list:
    features = get_user_features(user_id)
    embedding = json.loads(features["intent_embedding_json"])

    payload = {
        "query": current_query,
        "user_features": {
            "tool_calls_last_7d": int(features["tool_calls_last_7d"]),
            "avg_session_duration_s": float(features["avg_session_duration_s"]),
            "error_rate": float(features["error_rate_last_24h"]),
        },
        "user_embedding": embedding,
    }
    # invoke_endpoint(...)
    return payload

All feature values are returned as strings regardless of the declared FeatureType. Convert to the correct Python type (int(), float(), json.loads()) before passing to the model. The online store does not support batch reads natively — use batch_get_record for multi-entity lookups.

BatchGetRecord for multi-entity MCP tool calls

BatchGetRecord retrieves features for multiple entities across one or more feature groups in a single API call — up to 100 records per call, across up to 10 feature groups:

def get_features_for_session_participants(
    user_ids: list[str],
    tool_ids: list[str],
) -> dict:
    """
    Retrieve features for multiple users and tools in one batch call.
    Returns: {"users": {user_id: features}, "tools": {tool_id: features}}
    """
    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 record_batch in response["Records"]:
        feature_group = record_batch["FeatureGroupName"]
        record_id = record_batch["RecordIdentifier"]
        features = {f["FeatureName"]: f["ValueAsString"] for f in record_batch["Record"]}

        if feature_group == "mcp-user-behavior-features":
            result["users"][record_id] = features
        elif feature_group == "mcp-tool-popularity-features":
            result["tools"][record_id] = features

    # Entities in Errors had no record — handle cold start
    for error in response.get("Errors", []):
        entity_id = error["RecordIdentifier"]
        # log and fill defaults
        pass

    return result

Errors in the response contains entities that were not found — these are not raised as exceptions. Always check response["Errors"] and fill defaults for missing entities, especially in MCP tools where every user or document identifier may be new.

Writing features with PutRecord

MCP tools that take actions can write feature observations back to the store, keeping the feature values fresh for subsequent inference calls:

from datetime import datetime, timezone

def update_user_features_after_tool_call(
    user_id: str,
    tool_category: str,
    call_duration_s: float,
    success: bool,
) -> None:
    """Write updated behavioral features after an MCP tool execution."""
    # EventTime must be ISO 8601 in UTC — SageMaker rejects other formats
    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 updating
            # The online store merges with existing record; offline store appends a new row
        ],
        TargetStores=["OnlineStore"],  # write only to online store for low latency
        # Use TargetStores=["OnlineStore", "OfflineStore"] for both
    )

    # Fractional features with complex aggregation logic
    # Avoid computing rolling averages in the MCP tool — use a separate aggregation job
    # and write the precomputed aggregate periodically to the feature store

# Batch PutRecord is not supported — for bulk ingestion use the Data Wrangler
# export flow or a SageMaker Processing job with featurestore_runtime.put_record()
# in a worker pool

Partial PutRecord writes only update the features you include — the online store merges the new values with the existing record. The offline store always appends a new row (never updates in-place), so point-in-time queries over the offline store return the full history of feature values for each entity.

Querying the offline store via Athena

The offline store is a Glue catalog table backed by S3 Parquet or Iceberg files. MCP tools that serve analytics data or export training sets can query it via Athena:

import boto3
import time

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

ATHENA_DB = "sagemaker_featurestore"   # Glue database created by Feature Store
FEATURE_TABLE = "mcp_user_behavior_features"  # Glue table (group name with _ replacing -)
RESULTS_BUCKET = "mcp-athena-results"

def query_feature_history(user_id: str, since_date: str) -> list[dict]:
    """
    Query offline store for all feature observations for a user since a date.
    Returns rows in chronological order for training data generation.
    """
    query = f"""
    SELECT
        user_id,
        event_time,
        tool_calls_last_7d,
        avg_session_duration_s,
        error_rate_last_24h,
        write_time
    FROM "{ATHENA_DB}"."{FEATURE_TABLE}"
    WHERE user_id = '{user_id}'
      AND event_time >= '{since_date}'
      AND NOT is_deleted   -- Feature Store soft-deletes via is_deleted flag
    ORDER BY event_time ASC
    """

    # Start query
    execution = athena.start_query_execution(
        QueryString=query,
        QueryExecutionContext={"Database": ATHENA_DB},
        ResultConfiguration={
            "OutputLocation": f"s3://{RESULTS_BUCKET}/athena-results/",
        },
    )
    query_id = execution["QueryExecutionId"]

    # Poll until complete
    while True:
        status = athena.get_query_execution(QueryExecutionId=query_id)
        state = status["QueryExecution"]["Status"]["State"]
        if state in ("SUCCEEDED", "FAILED", "CANCELLED"):
            break
        time.sleep(1)

    if state != "SUCCEEDED":
        reason = status["QueryExecution"]["Status"].get("StateChangeReason", "unknown")
        raise RuntimeError(f"Athena query {state}: {reason}")

    # Paginate results
    results = []
    paginator = athena.get_paginator("get_query_results")
    pages = paginator.paginate(QueryExecutionId=query_id)
    headers = None
    for page in pages:
        for i, row in enumerate(page["ResultSet"]["Rows"]):
            values = [col.get("VarCharValue", "") for col in row["Data"]]
            if headers is None:
                headers = values
            else:
                results.append(dict(zip(headers, values)))
    return results

The Glue table created by Feature Store includes system columns: write_time (when the record was written to S3), api_invocation_time, and is_deleted. Always filter AND NOT is_deleted — Feature Store implements DeleteRecord as a soft delete that appends a row with is_deleted = true rather than removing the record from S3.

Monitor MCP servers that depend on Feature Store

SageMaker Feature Store online store has its own availability SLA. If the Feature Store endpoint degrades, every MCP tool call that depends on feature retrieval will fail or return stale defaults. AliveMCP probes each MCP endpoint continuously — catching Feature Store-induced failures before they cascade to users.

Join the waitlist →