MCP Tools for Amplitude — Event tracking, SDK vs HTTP API, throttling, and health monitoring

Amplitude's two delivery surfaces for server-side MCP tools

Amplitude's server-side integration architecture separates two concerns: event ingestion (high-throughput, low-latency write path) and user identity management (person profile properties, group associations). MCP tools that only capture events do not need the Identity API. MCP tools that sync user properties, cohort memberships, or group data also need the Identity API (/identify endpoint or dedicated Identify calls in the SDK).

import * as amplitude from '@amplitude/analytics-node';

// Initialize once at module level — creates one batched event queue
amplitude.init(process.env.AMPLITUDE_API_KEY, {
  // EU data residency: set serverUrl for EU endpoint
  // serverUrl: 'https://api.eu.amplitude.com/2/httpapi',
  flushIntervalMillis: 10_000,  // send queued events every 10 seconds
  flushQueueSize: 200,          // or when 200 events are queued, whichever comes first
  logLevel: amplitude.Types.LogLevel.Warn,
});

// Capture an event via SDK (automatic batching and retry)
function trackEvent(userId, deviceId, eventType, properties = {}) {
  amplitude.track(
    {
      event_type: eventType,
      user_id: userId,
      device_id: deviceId,    // at least one of user_id or device_id is required
      event_properties: properties,
      // Optional system properties:
      // app_version: '2.1.0',
      // platform: 'Node.js',
      // ip: '...',  Amplitude uses for geo-enrichment
      // time: Date.now(),  override event timestamp (default: now)
    }
    // Second arg: callback(error, response) — fires after flush, not immediately
  );
}

// Set user properties (Identity API via SDK)
function setUserProperties(userId, properties) {
  const identifyEvent = new amplitude.Identify();
  for (const [key, value] of Object.entries(properties)) {
    identifyEvent.set(key, value);          // set always (overwrite)
    // identifyEvent.setOnce(key, value);   // set only if not already set
    // identifyEvent.add(key, 1);           // increment a counter property
    // identifyEvent.append(key, value);    // append to array property
  }
  amplitude.identify(identifyEvent, { user_id: userId });
}

// CRITICAL: flush before process exit — unflushed events in the queue are lost
async function shutdown() {
  await amplitude.flush();
  // Note: @amplitude/analytics-node v1+ does NOT have a close() method
  // flush() is sufficient — it drains the queue and waits for all HTTP responses
}

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

Direct HTTP API v2 for synchronous delivery confirmation

The SDK's async batching is efficient but makes it impossible to confirm whether a specific event was accepted before returning a result to your MCP tool's caller. The HTTP API v2 gives synchronous confirmation: the response body includes events_ingested (how many were accepted) and server_upload_time (when Amplitude received them). Use the HTTP API directly when your MCP tool needs to confirm event delivery as part of a synchronous operation.

// Direct HTTP API v2 call — synchronous delivery confirmation
async function sendEventsHttp(apiKey, events) {
  const res = await fetch('https://api2.amplitude.com/2/httpapi', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ api_key: apiKey, events }),
    signal: AbortSignal.timeout(10_000),
  });

  const body = await res.json();

  if (res.status === 200) {
    // Success: { code: 200, events_ingested: N, payload_size_bytes: N, server_upload_time: epoch }
    return { accepted: body.events_ingested, uploadTime: body.server_upload_time };
  }

  if (res.status === 400) {
    // Validation error: events_with_invalid_fields, missing_field, silently_discarded_events
    // Amplitude may partially accept — events_ingested can be < events.length
    throw new Error(`Amplitude 400: ${body.error} — invalid fields: ${JSON.stringify(body.events_with_invalid_fields)}`);
  }

  if (res.status === 413) {
    // Payload too large — split into smaller batches (max 20MB or 2000 events per request)
    throw new Error(`Amplitude 413: payload too large — split into smaller batches`);
  }

  if (res.status === 429) {
    // THREE distinct throttle cases — check body to distinguish:
    if (body.exceeded_daily_quota?.user || body.exceeded_daily_quota?.device) {
      // Daily event quota exceeded for specific user_id or device_id
      // Do NOT retry with the same ID — the quota resets at midnight UTC
      throw new Error(`Amplitude 429: daily quota exceeded for users: ${JSON.stringify(body.exceeded_daily_quota)}`);
    }
    if (body.throttled_devices?.length || body.throttled_users?.length) {
      // Device or user sending too many events per second — wait 15s then retry
      const throttledIds = [...(body.throttled_devices || []), ...(body.throttled_users || [])];
      throw new Error(`Amplitude 429: per-device/user throttle — affected IDs: ${throttledIds.join(', ')}`);
    }
    // Project-level throttle — entire API key throttled, wait 30s
    throw new Error(`Amplitude 429: project-level throttle — back off 30s`);
  }

  if (res.status === 503) {
    // Server throttled — back off with exponential delay
    throw new Error(`Amplitude 503: server throttled — retry with exponential backoff`);
  }

  throw new Error(`Amplitude HTTP ${res.status}: ${JSON.stringify(body)}`);
}

// Build a well-formed Amplitude event object
function buildEvent(userId, eventType, properties = {}, opts = {}) {
  const now = Date.now();
  return {
    user_id: userId,
    // device_id: opts.deviceId,          // required if no user_id
    event_type: eventType,
    time: opts.time ?? now,               // epoch milliseconds
    event_properties: properties,
    user_properties: opts.userProperties ?? {},  // set per-event user properties
    groups: opts.groups ?? {},            // { "company": "Acme Corp" }
    app_version: opts.appVersion ?? process.env.npm_package_version,
    platform: opts.platform ?? 'Node',
    insert_id: opts.insertId ?? `${userId}-${eventType}-${now}`,  // deduplication key
  };
}

The insert_id field is Amplitude's deduplication key. Events with the same insert_id received within a 7-day window are deduplicated — only the first is stored. For retry logic (on 503 or network timeout), always preserve the original insert_id so that retried events don't create duplicates in Amplitude's event store.

Group analytics and cohort definitions

Amplitude's group analytics model links users to accounts, companies, or teams. Events can carry group membership properties, and group-level funnels let you analyze conversion at the account level rather than per user. MCP tools that serve multi-tenant applications should send group properties on every relevant event.

// Send an event associated with a group (e.g., company-level analytics)
async function trackGroupEvent(apiKey, userId, companyId, companyPlan, eventType, properties) {
  return sendEventsHttp(apiKey, [{
    user_id: userId,
    event_type: eventType,
    event_properties: properties,
    groups: {
      'company': companyId,          // group type: group value
    },
    group_properties: {
      '[Amplitude] Group': {
        'company': {
          plan: companyPlan,         // properties scoped to the group
          company_id: companyId,
        },
      },
    },
    insert_id: `${userId}-${eventType}-${Date.now()}`,
    time: Date.now(),
  }]);
}

Health monitoring for Amplitude integrations

Amplitude does not expose a programmatic health check endpoint. The status page at status.amplitude.com reflects operational status but not your project's ingestion rate or quota headroom. An MCP monitoring integration must probe the actual delivery path using a synthetic event to detect quota exhaustion, throttling, and ingestion pipeline issues.

async function probeAmplitude(apiKey) {
  const results = await Promise.allSettled([

    // 1. Delivery path — send a synthetic event and check response
    (async () => {
      const result = await sendEventsHttp(apiKey, [{
        user_id: '__alivemcp_probe__',
        event_type: '__health_probe__',
        time: Date.now(),
        insert_id: `probe-${Date.now()}`,
        event_properties: { synthetic: true },
      }]);
      if (result.accepted !== 1) {
        throw new Error(`Expected 1 event ingested, got ${result.accepted}`);
      }
      const lagMs = Date.now() - result.uploadTime;
      return { kind: 'delivery', accepted: true, serverLagMs: lagMs };
    })(),

    // 2. Batch endpoint health (separate from /2/httpapi — different infrastructure)
    (async () => {
      const res = await fetch('https://api2.amplitude.com/batch', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          api_key: apiKey,
          events: [{
            user_id: '__alivemcp_probe__',
            event_type: '__health_probe_batch__',
            time: Date.now(),
            insert_id: `probe-batch-${Date.now()}`,
          }],
        }),
        signal: AbortSignal.timeout(10_000),
      });
      // Batch endpoint returns 200 with {"code": 200} on success
      if (!res.ok) throw new Error(`Batch endpoint → ${res.status}`);
      const body = await res.json();
      if (body.code !== 200) throw new Error(`Batch endpoint code: ${body.code}`);
      return { kind: 'batch_endpoint', ok: true };
    })(),
  ]);

  const healthy = results.every(r => r.status === 'fulfilled');
  return {
    healthy,
    components: results.map(r =>
      r.status === 'fulfilled'
        ? { ...r.value, ok: true }
        : { ok: false, error: r.reason?.message }
    ),
  };
}

Set AliveMCP alert thresholds: critical when the HTTP API returns 429 with exceeded_daily_quota (all events for throttled users are silently discarded until midnight UTC); warning when the API returns 429 with per-device throttle (affects only specific IDs, not the whole project); info when server lag in the probe response exceeds 2 seconds (leading indicator of ingestion backpressure).

Common integration pitfalls

Missing device_id when no user_id is set

Amplitude requires at least one of user_id or device_id per event. Server-side MCP tools that process unauthenticated requests must set a stable device_id (e.g., a session ID) for pre-authentication events. Events missing both fields return 400 with missing_field: "user_id or device_id".

Not distinguishing the three 429 sub-cases

A generic "retry on 429" strategy is dangerous: retrying an event for a user whose daily quota is exceeded wastes API calls and delays other events. Parse the 429 response body — exceeded_daily_quota means skip that user until midnight UTC, while throttled_devices/throttled_users means back off 15 seconds and retry. Project-level throttle means back off 30+ seconds for all requests.

Treating 400 as a complete rejection

A 400 response can include events_ingested > 0 alongside silently_discarded_events > 0. Amplitude partially accepts batches where some events are valid and others are malformed. Log events_with_invalid_fields from the 400 response to identify the schema violation without losing the valid events.

Not calling flush() before Lambda/container shutdown

In serverless or containerized MCP servers, the process lifetime may end before the SDK's periodic flush timer fires. Always call await amplitude.flush() at the end of your request handler or in the shutdown hook — the SDK's in-memory queue is not persisted across process restarts.

Using /2/httpapi for high-volume ingestion

The /2/httpapi endpoint has lower rate limits than the /batch endpoint. For MCP tools that process high traffic (thousands of events per minute), use /batch instead. The batch endpoint is async (events may take minutes to appear in dashboards) but handles much higher throughput before hitting per-project throttle limits.

Related guides

• MCP tools for PostHog — event capture, feature flags, HogQL query API, and pipeline health
• MCP tools for Mixpanel — distinct_id model, people profiles, EU data residency, and ingestion health
• MCP tools for Segment — identify/track/group calls, Destination delivery opacity, and Protocols enforcement
• MCP tools for Heap Analytics — server-side event capture, identity linking, and retroactive data model
• MCP server health check patterns — composite endpoints and failure classification
• MCP server uptime monitoring — probes, intervals, and alert routing