Pinpoint Event Tracking for MCP Servers — PutEvents, Endpoint Attributes, and Kinesis Export

Custom event tracking with PutEvents

PutEventsCommand records one or more events against one or more endpoints in a single batch call. For MCP server tools, the natural places to record events are: when an alert is acknowledged by the user, when a user changes their notification preferences via a tool call, or when a user visits the status page linked in an alert.

import {
  PinpointClient,
  PutEventsCommand,
  EventsBatch,
  Event,
} from '@aws-sdk/client-pinpoint';

const client = new PinpointClient({ region: 'us-east-1' });
const APP_ID = process.env.PINPOINT_APP_ID!;

// Record a single custom event for an endpoint
async function recordEvent(
  endpointId: string,
  eventType: string,
  attributes?: Record,
  metrics?: Record
): Promise {
  const eventId = crypto.randomUUID();
  const event: Event = {
    EventType: eventType,       // dot-separated namespace convention: 'alert.acknowledged'
    Timestamp: new Date().toISOString(),
    Attributes: attributes,
    Metrics: metrics,
  };

  await client.send(new PutEventsCommand({
    ApplicationId: APP_ID,
    EventsRequest: {
      BatchItem: {
        [endpointId]: {
          // Events map — key is a client-assigned event ID (must be unique per batch item)
          Events: { [eventId]: event },
          // Endpoint update inline — avoids a separate UpdateEndpointCommand call
          // Omit if you don't need to update endpoint attributes alongside the event
          Endpoint: {},
        },
      },
    },
  }));
}

// MCP tool handler: user acknowledges a downtime alert
export async function acknowledgeAlert(
  userId: string,
  serverId: string,
  alertId: string
): Promise {
  const endpointId = `email-${userId}`;

  await recordEvent(
    endpointId,
    'alert.acknowledged',          // EventType — use a consistent naming convention
    {
      AlertId: alertId,
      ServerId: serverId,
      AckMethod: 'mcp_tool',       // string attributes for filtering in analytics
    },
    {
      TimeToAckSeconds: 142,       // numeric metrics for aggregation
    }
  );
}

// Batch events for multiple endpoints in one call (up to 100 events per request)
async function recordBatchEvents(
  events: Array<{ endpointId: string; type: string; attrs?: Record }>
): Promise {
  // Group events by endpointId since the BatchItem key is the endpoint ID
  const batchItem: Record = {};

  for (const ev of events) {
    if (!batchItem[ev.endpointId]) {
      batchItem[ev.endpointId] = { Events: {}, Endpoint: {} };
    }
    batchItem[ev.endpointId].Events![crypto.randomUUID()] = {
      EventType: ev.type,
      Timestamp: new Date().toISOString(),
      Attributes: ev.attrs,
    };
  }

  await client.send(new PutEventsCommand({
    ApplicationId: APP_ID,
    EventsRequest: { BatchItem: batchItem },
  }));
}

Event type naming: Pinpoint does not enforce a schema on event types, but a consistent dot-separated namespace (e.g., alert.sent, alert.acknowledged, user.preferences.changed) makes event filtering in Athena and the Pinpoint analytics console much cleaner. Avoid spaces in event type strings — they cause issues in Athena queries against the exported event stream.

Attribute and metric limits: Each event supports up to 40 string attributes (max 50 chars per key, 200 chars per value) and 40 numeric metrics (double precision). Attribute values must be strings — convert numbers or booleans to strings before including them in Attributes. Use Metrics for numeric data you plan to aggregate (sum, average, percentile).

Updating endpoint attributes inline with PutEvents

The BatchItem[endpointId].Endpoint field in PutEventsCommand allows updating endpoint attributes at the same time as recording an event, without a separate UpdateEndpointCommand call. This is useful for keeping endpoint attributes current as users interact with your MCP tools.

import { UpdateEndpointCommand, EndpointRequest } from '@aws-sdk/client-pinpoint';

// Record an event AND update endpoint attributes in one call
async function recordAlertAckWithEndpointUpdate(
  endpointId: string,
  userId: string,
  plan: string,
  lastActiveServer: string
): Promise {
  await client.send(new PutEventsCommand({
    ApplicationId: APP_ID,
    EventsRequest: {
      BatchItem: {
        [endpointId]: {
          Events: {
            [crypto.randomUUID()]: {
              EventType: 'alert.acknowledged',
              Timestamp: new Date().toISOString(),
              Attributes: { LastActiveServer: lastActiveServer },
            },
          },
          // Endpoint sub-object updates the endpoint's attributes
          // These attributes persist and can be used for segmentation
          Endpoint: {
            User: {
              UserId: userId,
              UserAttributes: {
                Plan: [plan],
                // Update 'LastAlertAck' timestamp for recency-based segments
                LastAlertAck: [new Date().toISOString()],
              },
            },
            Attributes: {
              ActiveServers: [lastActiveServer],
              NotificationPreference: ['email'],
            },
          },
        },
      },
    },
  }));
}

// Endpoint attributes drive dynamic segment membership
// Example: segment of users on the Team plan who have acknowledged an alert
// in the last 30 days (LastAlertAck attribute used as filter)
// This segment can be targeted for a "new feature" campaign or renewal reminder

Endpoint attributes set via the Endpoint sub-object in PutEventsCommand behave identically to attributes set via UpdateEndpointCommand — they are merged with existing attributes (not replaced), persisted in Pinpoint's endpoint store, and immediately available for segment evaluation and targeting.

Attribute cardinality: Avoid high-cardinality values in endpoint attributes that you plan to use for segment filters — values like UUIDs, timestamps, or server names from an open-ended set make it impossible to create useful segment filters. Store high-cardinality values only in event attributes, not on the endpoint itself.

Automatic delivery event types

Pinpoint automatically captures delivery events for messages sent through its channels. These events appear in the Pinpoint analytics console and in the event stream without any extra instrumentation.

Configuring the event stream to Kinesis Firehose

The Pinpoint analytics console provides aggregated metrics (open rates, click rates, delivery rates per campaign), but for custom queries — which servers trigger the most alerts, what percentage of users acknowledge within 15 minutes, alert fatigue analysis — you need the raw event stream. Pinpoint's event stream exports every event (custom and automatic) to a Kinesis Firehose delivery stream in near-real-time.

import { PutEventStreamCommand } from '@aws-sdk/client-pinpoint';

// Configure the event stream (run once during setup)
// Requires:
// 1. A Kinesis Firehose delivery stream in the same region
// 2. An IAM role that Pinpoint can assume to put records to Firehose
async function configureEventStream(
  firehoseStreamArn: string,
  roleArn: string
): Promise {
  await client.send(new PutEventStreamCommand({
    ApplicationId: APP_ID,
    WriteEventStream: {
      DestinationStreamArn: firehoseStreamArn,
      // IAM role with kinesis-firehose:PutRecord + kinesis-firehose:PutRecordBatch
      // on the specific stream ARN
      RoleArn: roleArn,
    },
  }));
}

// IAM trust policy for the role (Pinpoint must be able to assume it):
// {
//   "Version": "2012-10-17",
//   "Statement": [{
//     "Effect": "Allow",
//     "Principal": { "Service": "pinpoint.amazonaws.com" },
//     "Action": "sts:AssumeRole"
//   }]
// }
//
// IAM permissions policy on the role:
// {
//   "Version": "2012-10-17",
//   "Statement": [{
//     "Effect": "Allow",
//     "Action": ["firehose:PutRecord", "firehose:PutRecordBatch"],
//     "Resource": "arn:aws:firehose:REGION:ACCOUNT:deliverystream/STREAM_NAME"
//   }]
// }

Once the event stream is configured, Kinesis Firehose buffers the events and delivers them to S3 in GZIP-compressed newline-delimited JSON files. The default buffer size is 5 MB or 300 seconds, whichever comes first. Each record is one JSON object per line:

// Sample event stream record (one line in the NDJSON file)
{
  "event_type": "alert.acknowledged",
  "event_timestamp": 1731124800000,
  "arrival_timestamp": 1731124802341,
  "event_version": "3.1",
  "application": { "app_id": "abc123", "sdk": {} },
  "client": { "client_id": "email-user-456" },
  "device": {},
  "session": {},
  "attributes": {
    "AlertId": "alert-789",
    "ServerId": "my-mcp-server",
    "AckMethod": "mcp_tool"
  },
  "metrics": {
    "TimeToAckSeconds": 142
  },
  "awsAccountId": "123456789012"
}

The S3 prefix follows the pattern events/YYYY/MM/DD/HH/. Use Athena to query the data with a schema that matches the event stream format. The Pinpoint documentation provides a sample Athena DDL — the key columns are event_type, event_timestamp (epoch milliseconds, cast with from_unixtime(event_timestamp / 1000)), and the nested attributes and metrics maps.

Practical analytics patterns for MCP notification tools

Four queries that answer the questions MCP server notification tools most need answered:

-- 1. Alert acknowledgment rate by server (last 7 days)
-- How many alerts per server are being acknowledged vs silently missed?
SELECT
  attrs['ServerId'] AS server_id,
  COUNT_IF(event_type = 'alert.sent') AS alerts_sent,
  COUNT_IF(event_type = 'alert.acknowledged') AS alerts_acked,
  ROUND(
    100.0 * COUNT_IF(event_type = 'alert.acknowledged')
      / NULLIF(COUNT_IF(event_type = 'alert.sent'), 0),
    1
  ) AS ack_rate_pct
FROM pinpoint_events
WHERE event_timestamp > (UNIX_TIMESTAMP(NOW() - INTERVAL 7 DAY) * 1000)
  AND event_type IN ('alert.sent', 'alert.acknowledged')
GROUP BY attrs['ServerId']
ORDER BY ack_rate_pct ASC;  -- lowest ack rates first (alert fatigue candidates)

-- 2. Time-to-acknowledge distribution (p50, p90, p99)
SELECT
  APPROX_PERCENTILE(CAST(metrics['TimeToAckSeconds'] AS double), 0.5)  AS p50_seconds,
  APPROX_PERCENTILE(CAST(metrics['TimeToAckSeconds'] AS double), 0.9)  AS p90_seconds,
  APPROX_PERCENTILE(CAST(metrics['TimeToAckSeconds'] AS double), 0.99) AS p99_seconds
FROM pinpoint_events
WHERE event_type = 'alert.acknowledged'
  AND event_timestamp > (UNIX_TIMESTAMP(NOW() - INTERVAL 30 DAY) * 1000);

-- 3. Email open rate by alert type (were critical alerts opened faster?)
SELECT
  attributes['AlertSeverity'] AS severity,
  COUNT_IF(event_type = '_email.send') AS sent,
  COUNT_IF(event_type = '_email.opened') AS opened,
  ROUND(100.0 * COUNT_IF(event_type = '_email.opened')
    / NULLIF(COUNT_IF(event_type = '_email.send'), 0), 1) AS open_rate_pct
FROM pinpoint_events
WHERE event_timestamp > (UNIX_TIMESTAMP(NOW() - INTERVAL 30 DAY) * 1000)
GROUP BY attributes['AlertSeverity'];

-- 4. Opt-out rate by notification channel (which channel loses users fastest?)
SELECT
  DATE_TRUNC('week', from_unixtime(event_timestamp / 1000)) AS week,
  CASE
    WHEN event_type = '_email.complaint' THEN 'email'
    WHEN event_type = '_SMS.OPT_OUT' THEN 'sms'
  END AS channel,
  COUNT(*) AS opt_outs
FROM pinpoint_events
WHERE event_type IN ('_email.complaint', '_SMS.OPT_OUT')
GROUP BY 1, 2
ORDER BY week DESC;

The column names in the Athena schema use underscore-separated names (event_type, event_timestamp) while the raw JSON uses dot-notation (event.type). The Pinpoint-provided Athena DDL normalizes these for you — use it rather than manually mapping field names.

Common errors and their causes