Kinesis Data Analytics for MCP Servers

Kinesis Data Analytics Studio vs Managed Flink — which to use

Studio notebooks (Apache Zeppelin + Flink) are designed for interactive exploration — write Flink SQL in a notebook cell, see results in seconds, iterate on window definitions and filters. They're not designed for 24/7 production jobs: notebooks have higher per-KPU costs, checkpointing is limited, and you can't version-control a notebook as cleanly as a Flink JAR. Start with Studio for exploration, graduate to a Managed Flink application JAR for production.

Flink SQL windowed queries for MCP tool call metrics

The most common MCP server analytics query: "how many calls per tool per tenant in the last 5 minutes, and what fraction failed?" In Flink SQL this is a tumbling window aggregation over the input Kinesis stream.

-- Flink SQL for Kinesis Data Analytics Studio
-- Create source table connected to Kinesis Data Stream
CREATE TABLE mcp_tool_calls (
  tenant_id     VARCHAR,
  tool_name     VARCHAR,
  session_id    VARCHAR,
  success       BOOLEAN,
  duration_ms   BIGINT,
  ts            TIMESTAMP(3),
  WATERMARK FOR ts AS ts - INTERVAL '10' SECOND  -- allow 10s late arrivals
) WITH (
  'connector' = 'kinesis',
  'stream' = 'mcp-tool-events',
  'aws.region' = 'us-east-1',
  'scan.stream.initpos' = 'LATEST',
  'format' = 'json',
  'json.timestamp-format.standard' = 'ISO-8601'
);

-- Tumbling 5-minute window — tool call rate and error rate per tenant
SELECT
  tenant_id,
  tool_name,
  TUMBLE_START(ts, INTERVAL '5' MINUTE) AS window_start,
  TUMBLE_END(ts, INTERVAL '5' MINUTE)   AS window_end,
  COUNT(*)                               AS total_calls,
  SUM(CASE WHEN NOT success THEN 1 ELSE 0 END) AS failed_calls,
  CAST(SUM(CASE WHEN NOT success THEN 1 ELSE 0 END) AS DOUBLE)
    / COUNT(*) * 100                     AS error_rate_pct,
  AVG(CAST(duration_ms AS DOUBLE))       AS avg_duration_ms,
  MAX(duration_ms)                       AS p100_duration_ms
FROM mcp_tool_calls
GROUP BY
  tenant_id,
  tool_name,
  TUMBLE(ts, INTERVAL '5' MINUTE);

-- Sliding window: 5-minute window advancing every 1 minute (for smoother trend lines)
SELECT
  tenant_id,
  HOP_START(ts, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE) AS window_start,
  COUNT(*) AS calls
FROM mcp_tool_calls
GROUP BY
  tenant_id,
  HOP(ts, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE);

-- Session window: group calls within the same session (gap closes after 30s of inactivity)
SELECT
  session_id,
  SESSION_START(ts, INTERVAL '30' SECOND) AS session_start,
  SESSION_END(ts, INTERVAL '30' SECOND)   AS session_end,
  COUNT(*)                                 AS tools_used,
  COUNT(DISTINCT tool_name)                AS unique_tools
FROM mcp_tool_calls
GROUP BY session_id, SESSION(ts, INTERVAL '30' SECOND);

Event time vs processing time windows

Flink supports two notions of time for window operations, and choosing the wrong one produces subtly wrong metrics for MCP event streams:

Processing time — windows are based on when Flink processes the record, not when the event occurred. A 5-minute tumbling window fires every 5 minutes of wall clock time. Late-arriving records (due to network delays, client buffering) are included in whatever window happens to be open when they arrive. Simpler and has lower latency, but your metrics can include events from several minutes ago mixed with current events.

Event time — windows are based on the event's own timestamp (ts field). Flink uses a watermark to track progress: the watermark is the highest event timestamp seen minus a late-arrival tolerance. A window fires only when the watermark advances past the window's end time. This gives accurate temporal attribution but adds latency equal to the late-arrival tolerance. For MCP server metrics where events are generated locally and arrive with low skew (sub-second network hops), an event time window with 10-second late tolerance is accurate and adds only 10 seconds of latency.

Use processing time for: operational alerting where low latency matters more than perfect attribution. Use event time for: billing, audit reports, or any metric where you need to know "what happened between 14:00 and 14:05" accurately.

Random Cut Forest anomaly detection on MCP event streams

Amazon Kinesis Data Analytics for SQL (the deprecated predecessor to Managed Flink) had a built-in RANDOM_CUT_FOREST SQL function. In Managed Flink, you implement RCF anomaly detection using the Flink ML library or the Amazon Random Cut Forest Java library. RCF is particularly well-suited for MCP tool call anomalies because it detects multi-dimensional anomalies (unusual combination of tool_name + duration_ms + error rate) without needing labeled training data.

// Managed Flink Java job — Random Cut Forest anomaly detection
// Dependencies: flink-streaming-java, software.amazon.randomcutforest:randomcutforest

import com.amazon.randomcutforest.RandomCutForest;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class McpAnomalyDetector {
  private static final int SHINGLE_SIZE = 8;     // window of recent samples per key
  private static final double ANOMALY_THRESHOLD = 3.0; // z-score threshold

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.enableCheckpointing(60_000); // 60-second checkpoint interval

    DataStream<ToolCallEvent> stream = env
      .addSource(new FlinkKinesisConsumer<>("mcp-tool-events", new ToolCallEventSchema(), kinesisConsumerConfig));

    // Key by tenant to isolate per-tenant anomaly detection
    stream
      .keyBy(event -> event.getTenantId())
      .flatMap(new RCFAnomalyFunction())
      .filter(result -> result.getAnomalyScore() > ANOMALY_THRESHOLD)
      .addSink(new FlinkKinesisProducer<>("mcp-anomalies", new AnomalySchema(), kinesisProducerConfig));

    env.execute("MCP Tool Call Anomaly Detection");
  }

  // Stateful FlatMapFunction — RCF model is Flink state, checkpointed to S3
  private static class RCFAnomalyFunction
      extends RichFlatMapFunction<ToolCallEvent, AnomalyResult> {

    private transient RandomCutForest forest;
    private transient ValueState<Long> recordCount;

    @Override
    public void open(Configuration parameters) {
      // Build RCF with dimensions: [duration_ms, error_flag, hour_of_day]
      // Warm-up: first 256 records train the model without emitting scores
      forest = RandomCutForest.builder()
        .dimensions(3)
        .sampleSize(256)
        .numberOfTrees(50)
        .build();

      recordCount = getRuntimeContext().getState(
        new ValueStateDescriptor<>("count", Long.class, 0L)
      );
    }

    @Override
    public void flatMap(ToolCallEvent event, Collector<AnomalyResult> out) throws Exception {
      double[] point = {
        event.getDurationMs(),
        event.isSuccess() ? 0.0 : 1.0,
        event.getTimestamp().getHour(),
      };

      double score = forest.getAnomalyScore(point);
      forest.update(point);

      long count = recordCount.value();
      recordCount.update(count + 1);

      // Only emit scores after warm-up period (256 records builds baseline)
      if (count > 256) {
        out.collect(new AnomalyResult(
          event.getTenantId(),
          event.getToolName(),
          score,
          event.getTimestamp()
        ));
      }
    }
  }
}

Stream enrichment with S3 reference data

MCP tool call events typically contain IDs but not descriptive metadata — tenant IDs without tier/plan information, tool names without category labels, session IDs without user metadata. Enrich the stream at processing time by loading reference data from S3 into Flink state at startup, then refreshing it periodically.

-- Flink SQL enrichment using a lookup join with an S3-backed dimension table
-- The S3 file is a JSON or CSV lookup table refreshed from your application DB

CREATE TABLE tenant_metadata (
  tenant_id   VARCHAR,
  plan_tier   VARCHAR,  -- 'free' | 'author' | 'team' | 'enterprise'
  region      VARCHAR,
  PRIMARY KEY (tenant_id) NOT ENFORCED
) WITH (
  'connector' = 'filesystem',
  'path' = 's3://mcp-reference-data/tenant-metadata/current.json',
  'format' = 'json'
);

-- Lookup join: enrich each event with tenant plan tier
-- Temporal join adds processing-time enrichment (uses latest version of dimension table)
SELECT
  e.tenant_id,
  e.tool_name,
  e.duration_ms,
  e.success,
  m.plan_tier,
  m.region
FROM mcp_tool_calls AS e
LEFT JOIN tenant_metadata FOR SYSTEM_TIME AS OF e.proctime AS m
  ON e.tenant_id = m.tenant_id;

Cost model — KPU sizing and idle cost trap

Kinesis Data Analytics bills per Kinesis Processing Unit (KPU). One KPU = 1 vCPU + 4 GB RAM. As of 2026, a KPU costs approximately $0.11/hour. Managed Flink applications minimum 1 KPU; scale based on throughput and operator parallelism. A common mistake: leaving a Managed Flink application in RUNNING state while it's idle still bills full KPU-hours. Use StopApplication to stop billing when not in use (state is preserved in S3); use StartApplication with FlinkRunConfiguration.AllowNonRestoredState: false to resume from the last checkpoint.

# Stop application — billing stops, state preserved in S3
aws kinesisanalyticsv2 stop-application \
  --application-name mcp-analytics \
  --force

# Start from last checkpoint (false = fail if state schema changed)
aws kinesisanalyticsv2 start-application \
  --application-name mcp-analytics \
  --run-configuration '{
    "FlinkRunConfiguration": {
      "AllowNonRestoredState": false
    }
  }'

# Describe current application state and KPU count
aws kinesisanalyticsv2 describe-application \
  --application-name mcp-analytics \
  --query 'ApplicationDetail.{Status:ApplicationStatus,KPUs:ApplicationConfigurationDescription.FlinkApplicationConfigurationDescription.ParallelismConfigurationDescription.Parallelism}'

Common failure modes reference