Amplitude webhook adapter

Availability: Available

Amplitude: Create the source through the API as shown below. The console source catalog does not list this adapter yet.

Stream your Amplitude events into Tessera with zero code. Your product analytics taxonomy is yours — Song Played, Add To Cart, whatever you named it — so this adapter does not enumerate event types: every Amplitude event becomes a namespaced canonical event (Song Played → amplitude.song_played) about the user (or, before login, the device) who did it.

Prerequisites

• An API key with the admin scope.
• An Amplitude project where you can add an event-streaming Webhook destination.

How authentication works

Amplitude does not sign webhook payloads, so the receiver authenticates Amplitude to Tessera with a shared credential sent over TLS: every delivery must carry Authorization: Bearer <webhook_secret>, compared in constant time against the source's secret before the body is parsed. The secret is envelope-encrypted with Cloud KMS when the source is created and is never returned by any API: reads show {"set": true} in its place. Transport security is TLS plus this credential; there is no payload HMAC to verify. Rotating the secret is a source update and takes effect within 60 seconds.

Create the source

Generate a long random secret, then create a kind=stream source with the adapter subtype:

{
  "name": "amplitude-events",
  "kind": "stream",
  "config": {
    "subtype": "adapter",
    "provider": "amplitude",
    "webhook_secret": "REPLACE_WITH_A_LONG_RANDOM_SECRET"
  }
}

The source id (src_…) in the response is the last path segment of your receiver URL:

https://ingest.staging.tessera.zone/v1/adapters/amplitude/{src_id}

Configure the Amplitude webhook destination

In Amplitude (Data → Destinations → Webhook, event streaming):

1. Set the URL to your receiver URL.
2. Add the header Authorization with the value Bearer <webhook_secret>.
3. Select the events to forward (forwarding everything is fine: the catalog maps any event name).
4. Keep the event payload in the shape below. This is the payload template the v1 catalog is pinned to — Amplitude's event fields, one JSON object per event (a JSON array or an {"events": [...]} wrapper of them also works):

amplitude-delivery.json

[
  {
    "event_type": "Song Played",
    "user_id": "listener-20417",
    "device_id": "dev-7c1e9a40-ios",
    "time": 1759312800000,
    "insert_id": "amp-insert-5f2b8d10-0001",
    "session_id": 1759312000000,
    "event_properties": {"genre": "jazz", "track_id": "trk-00412", "duration_ms": 214000},
    "user_properties": {"plan": "premium"},
    "platform": "iOS",
    "os_name": "ios",
    "os_version": "18.6",
    "device_brand": "Apple",
    "device_model": "iPhone16,2",
    "country": "Canada",
    "region": "Quebec",
    "city": "Montreal",
    "language": "French",
    "ip": "192.0.2.10",
    "library": "amplitude-swift/1.9.2"
  },
  {
    "event_type": "Song Skipped",
    "user_id": "listener-20417",
    "device_id": "dev-7c1e9a40-ios",
    "time": 1759313014500,
    "insert_id": "amp-insert-5f2b8d10-0002",
    "event_properties": {"track_id": "trk-00413", "position_ms": 12000},
    "platform": "iOS",
    "library": "amplitude-swift/1.9.2"
  }
]

Download connect/amplitude/amplitude-delivery.json — this exact file runs in CI on every change (hosts shown are for this environment).

The fields the catalog reads: event_type (the event name), user_id and device_id (identity), time (epoch milliseconds), insert_id (deduplication), event_properties and user_properties (attributes), and platform, os_name, os_version, device_brand, device_model, country, region, city, language, ip, library (context). A template that renames or reshapes any of these needs a mapping override (below) — otherwise the affected events are dead-lettered, not guessed.

Event coverage

Generated from the released amplitude mapping catalog — the catalog file is the single source of truth:

Catalog version v1 (new sources default to the latest released version; existing sources pin the version they were created with).

Passthrough mapping — every provider event that is not denied and has no explicit entry maps by this rule (<name> is the provider event name through the sanitizer: lowercased, other characters → _, ≤ 64 characters):

Field	Mapping
Canonical event_type	amplitude.<name>
Entity	user ← /body/user_id, else device ← /body/device_id
Secondary entities	device ← /body/device_id
Timestamp	/body/time (unix_ms)
Event id	/body/insert_id
Attributes	members of /body/event_properties; members of /body/user_properties (keys prefixed user_)
Context	city ← /body/city, country ← /body/country, device_brand ← /body/device_brand, device_model ← /body/device_model, ip ← /body/ip, language ← /body/language, library ← /body/library, os_name ← /body/os_name, os_version ← /body/os_version, platform ← /body/platform, region ← /body/region
Denied provider events	$identify, $merge_user, $groupidentify

Pointers resolve against the delivery document {"headers": {…lowercased…}, "body": {…payload…}}; the event-type discriminator is /body/event_type.

What that means for your data:

• Event types. The name goes through one sanitizer: lowercased, every run of other characters becomes _, trimmed, prefixed with e_ if it does not start with a letter, at most 64 characters. Two names that sanitize identically (Song Played, song-played) share one canonical type; the collision is counted, and an explicit entry separates them.
• Identity. user_id is the entity (user); without it, device_id is (device). With both, the device is a secondary entity. Identifiers (user_id, device_id, amplitude_id and the like, at any level) never appear as attributes or context — identity is never a feature.
• Attributes. event_properties map verbatim; user_properties map with a user_ prefix. One level of nested object flattens ({"cart": {"total": 10}} → cart_total); deeper nesting and values the event model cannot hold are dropped and counted (adapter_attr_dropped_total). Numbers pass through exactly as sent.
• Denied by default. Identity-management events ($identify, $merge_user, $groupidentify) are acknowledged and dropped — Tessera never merges identities.
• New types are governed. Adapter-produced event types and attributes arrive as draft contract changes: they widen intake, not model input, until you review them.

Retries and idempotency

insert_id becomes the canonical event_id, so Amplitude's redeliveries are no-ops downstream. Without an insert_id the event is published without one and the platform derives a deterministic id, so identical redeliveries still deduplicate. The receiver answers 200 only after every event in the request is durably queued; if any publish fails it answers 503 so Amplitude retries the whole request (safe, by the same ids). One request carries at most 500 events / 5 MiB.

Events that cannot be mapped — no user_id and no device_id, a missing time, time: 0, or a time more than a day in the future — are parked in the dead-letter view with reason=mapping_failed (the request still answers 200) and can be replayed after you fix the template or add an override.

Customize the mapping

The standard source-mapping routes serve this adapter:

• GET /v1/sources/{src_id}/mapping — the resolved catalog and your overrides.
• PUT /v1/sources/{src_id}/mapping — overrides, validated field by field:
  • an explicit entry for one event name beats the passthrough rule, e.g. map Song Played to your own track.played with chosen attributes;
  • {"enabled": false} for an event name suppresses it;
  • a field-wise passthrough override changes the shared rule — for example "deny_source_events": [] to stop denying identity events, or extra context_fields.
• POST /v1/sources/{src_id}/mapping:preview with {"sample_payload": …} — runs the merged mapping on a sample delivery (object, array or {"events": [...]}) and returns exactly the events the live path would publish, with zero side effects (requires admin).

Backfill history

Push history through the same receiver: export events with Amplitude's Export API, convert each line to the live payload shape (Amplitude's export files carry event_time and $insert_id; the helper below rewrites those two fields and nothing else), and POST the result as NDJSON (Content-Type: application/x-ndjson) in chunks of at most 500 events with the same Bearer credential:

export_to_ndjson.py

"""Backfill helper: Amplitude Export API lines -> <=500-event NDJSON chunks.

Amplitude's export files carry ``event_time`` ("YYYY-MM-DD HH:MM:SS.ffffff",
UTC) and ``$insert_id``; the receiver's v1 catalog reads the live-stream
shape (``time`` in epoch milliseconds, ``insert_id``). This rewrites those
two fields and nothing else, then writes chunk files of at most 500 events.

Usage: python3 export_to_ndjson.py <export.jsonl> <out-dir>
"""

import json
import sys
from datetime import UTC, datetime
from pathlib import Path

CHUNK = 500  # the receiver's per-request cap


def normalize(event: dict) -> dict:
    out = dict(event)
    if "time" not in out and "event_time" in out:
        parsed = datetime.strptime(out.pop("event_time"), "%Y-%m-%d %H:%M:%S.%f")
        out["time"] = int(parsed.replace(tzinfo=UTC).timestamp() * 1000)
    if "insert_id" not in out and "$insert_id" in out:
        out["insert_id"] = out.pop("$insert_id")
    return out


def main(source: str, out_dir: str) -> None:
    lines = [line for line in Path(source).read_text().splitlines() if line.strip()]
    events = [normalize(json.loads(line)) for line in lines]
    target = Path(out_dir)
    target.mkdir(parents=True, exist_ok=True)
    for index in range(0, len(events), CHUNK):
        chunk = events[index : index + CHUNK]
        body = "".join(json.dumps(event, separators=(",", ":")) + "\n" for event in chunk)
        (target / f"chunk-{index // CHUNK:05d}.ndjson").write_text(body)
    print(f"backfill: {len(events)} events in {(len(events) + CHUNK - 1) // CHUNK} chunk(s)")


if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2])

Download connect/amplitude/export_to_ndjson.py — this exact file runs in CI on every change (hosts shown are for this environment).

Backfilled events carry their original insert_id, so overlapping a backfill with live streaming is harmless.

Test and verify

This sample (run in CI against the local stack) creates the source, sends a delivery, shows the denied identity event and a rejected credential, previews the mapping, backfills an export file and reads the timeline:

01-receive-and-backfill.sh

#!/usr/bin/env bash
# Amplitude webhook adapter: create the source, receive deliveries, preview
# the mapping, then backfill exported history through the same receiver.
set -euo pipefail

: "${TESSERA_API_BASE:?e.g. https://api.staging.tessera.zone/v1}"
: "${TESSERA_INGEST_BASE:?e.g. https://ingest.staging.tessera.zone/v1}"
: "${TESSERA_API_KEY:?your tsk_test_… key}"
AUTH=(-H "Authorization: Bearer $TESSERA_API_KEY")

summarize() {
  python3 -c 'import json, sys; r = json.load(sys.stdin); print("%s: %s events=%s" % (sys.argv[1], r["outcome"], r["events"]))' "$1"
}

# 1. A random shared credential. Paste the same value into Amplitude's
#    webhook destination as the Bearer token header.
WEBHOOK_SECRET="$(python3 -c 'import secrets; print(secrets.token_urlsafe(32))')"

# 2. Create the adapter source. The secret is write-only and sealed with
#    Cloud KMS before it is stored; reads show {"set": true} in its place.
SOURCE="$(curl -sS -X POST "$TESSERA_API_BASE/sources" "${AUTH[@]}" \
  -H "Content-Type: application/json" \
  -d '{"name": "amplitude-events", "kind": "stream",
       "config": {"subtype": "adapter", "provider": "amplitude",
                  "webhook_secret": "'"$WEBHOOK_SECRET"'"}}')"
SRC_ID="$(printf '%s' "$SOURCE" | python3 -c 'import json, sys; print(json.load(sys.stdin)["source_id"])')"
RECEIVER="$TESSERA_INGEST_BASE/adapters/amplitude/$SRC_ID"
echo "receiver: $TESSERA_INGEST_BASE/adapters/amplitude/{src_id}"

# 3. A delivery exactly as Amplitude's webhook destination sends it: a JSON
#    array of events with the shared credential as a Bearer token.
curl -sS -X POST "$RECEIVER" --oauth2-bearer "$WEBHOOK_SECRET" \
  -H "Content-Type: application/json" --data-binary @amplitude-delivery.json \
  | summarize delivery

# Identity-management events ($identify, $merge_user, $groupidentify) are
# denied by the catalog: acknowledged, counted, never published.
curl -sS -X POST "$RECEIVER" --oauth2-bearer "$WEBHOOK_SECRET" \
  -H "Content-Type: application/json" --data-binary @amplitude-identify.json \
  | summarize identify

# A wrong credential is rejected before the body is even parsed.
WRONG_SECRET="$(python3 -c 'import secrets; print(secrets.token_urlsafe(16))')"
STATUS="$(curl -sS -o /dev/null -w '%{http_code}' -X POST "$RECEIVER" \
  --oauth2-bearer "$WRONG_SECRET" \
  -H "Content-Type: application/json" --data-binary @amplitude-delivery.json)"
echo "wrong credential: $STATUS"

# 4. Preview: the exact canonical events the live path publishes, with zero
#    side effects.
curl -sS -X POST "$TESSERA_API_BASE/sources/$SRC_ID/mapping:preview" "${AUTH[@]}" \
  -H "Content-Type: application/json" \
  -d "{\"sample_payload\": $(cat amplitude-delivery.json)}" \
  | python3 -c 'import json, sys; r = json.load(sys.stdin); print("preview:", " ".join(e["event_type"] for e in r["events"]))'

# 5. Backfill: Amplitude Export API lines -> NDJSON chunks of <= 500 events,
#    posted to the same receiver with the same credential.
CHUNKS="$(mktemp -d)"
python3 export_to_ndjson.py amplitude-export.jsonl "$CHUNKS"
for chunk in "$CHUNKS"/chunk-*.ndjson; do
  curl -sS -X POST "$RECEIVER" --oauth2-bearer "$WEBHOOK_SECRET" \
    -H "Content-Type: application/x-ndjson" --data-binary @"$chunk" \
    | summarize backfill-chunk
done

# 6. The published events land on the listener's timeline.
COUNT=0
for _ in $(seq 1 30); do
  COUNT="$(curl -sS "$TESSERA_API_BASE/entities/user/listener-20417/events" "${AUTH[@]}" \
    | python3 -c 'import json, sys; print(len(json.load(sys.stdin)["data"]))')"
  if [ "$COUNT" -ge 2 ]; then break; fi
  sleep 1
done
echo "timeline events for listener-20417: $COUNT"

Download connect/amplitude/01-receive-and-backfill.sh — this exact file runs in CI on every change (hosts shown are for this environment).

Troubleshooting

Symptom	Cause	Fix
401 on deliveries	missing or wrong Authorization: Bearer header	copy the source's secret into the destination header exactly (no quotes or spaces)
404 on deliveries	wrong src_id, a disabled source, or a Mixpanel source behind an Amplitude URL	re-check the receiver URL against the source
400 invalid_argument	unparseable body, more than 500 events, or a non-object member	fix the template; split backfills into ≤ 500-event chunks
events parked as mapping_failed	no user_id/device_id, or time missing, zero, not epoch milliseconds, or in the future	fix the template, or override timestamp_format; then replay from the dead-letter view
an attribute is missing	nested more than one level, or not a valid attribute value	flatten it in the template, or map it with an explicit entry