metricflow-ai 0.1.0

Server-side Python SDK for MetricFlow analytics.

metricflow-ai

Server-side Python SDK for MetricFlow analytics.

Table of Contents

• Installation
• Requirements
• Quick Start
• Identity Resolution
• Configuration
• Core API
• User Profiles (people_*)
• Consumers
• Revenue Tracking
• Error Handling
• License

---

Installation

pip install metricflow-ai

Requirements

• Python >= 3.9
• A MetricFlow client_id + client_secret (server-side credential — never expose client_secret to a browser)

---

Quick Start

from metricflow_ai import MetricFlow

mf = MetricFlow(
    client_id="app_123",
    client_secret="secret_123",
    # Defaults to https://d1kmmbafv229pu.cloudfront.net
)

mf.track("user_123", "payment_succeeded", {"amount": 99})
mf.identify("user_123", traits={"plan": "pro"})
mf.revenue.payment_succeeded(
    user_id="user_123",
    event_id="stripe_evt_123",
    amount=99,
    currency="USD",
    provider="stripe",
    payment_id="pi_123",
    invoice_id="in_123",
    subscription_id="sub_123",
    plan_id="pro",
)
mf.people_set("user_123", {"plan": "pro"})
mf.flush()

Backend SDKs are stateless by design. Pass an identity on every event. For request-linked backend events, pass request context so MetricFlow can enrich IP, browser, OS, and device from the original visitor request.

Use track() for normal backend custom events such as workspace_created or ai_summary_generated. Use revenue.* helpers for money/subscription events confirmed by your backend or payment webhook. Revenue helpers require event_id so repeated webhook deliveries produce the same insert_id and idempotency_key for ingestion deduplication.

---

Identity Resolution

track()'s first argument (distinct_id) is required. Internally, identity resolves in this order:

1. meta["user_id"] / properties["user_id"] — the strongest identifier.
2. The distinct_id argument itself — used as user_id only if neither of the above is set.
3. meta["anonymous_id"] / properties["anonymous_id"] — for events with no known user.

Linking a backend event to a user from your frontend SDK

To attribute a backend event (e.g. a payment webhook firing revenue.payment_succeeded) to the same person already tracked by your browser or React Native app, pass the exact same user ID that SDK used in its own identify() call:

# Browser or React Native app, at login:
#   metricflow.identify("user_123", { email: "...", plan: "free" })

# Backend, later — same user id, different SDK:
mf.revenue.payment_succeeded(
    user_id="user_123",  # must match the frontend identify() call exactly
    event_id="stripe_evt_123",
    amount=99,
    currency="USD",
    provider="stripe",
)

Both events resolve to the same user profile automatically — no extra field, session ID, or setup is needed. User IDs are compared exactly (case-insensitive only when the value looks like an email); IDs like usr_abc123 must match byte-for-byte across every SDK sending them.

---

Configuration

mf = MetricFlow(
    client_id="app_123",
    client_secret="secret_123",
    api_host="https://d1kmmbafv229pu.cloudfront.net",
    request_timeout=90,
    retry_limit=4,
    retry_backoff_factor=0.25,
)

Field	Type	Default	Description
client_id	str	—	Required. Your project's client ID.
client_secret	str	—	Required. Your project's server-side secret. Never expose this in browser code.
api_host	str	https://d1kmmbafv229pu.cloudfront.net	Base API host.
endpoint	str	—	Override the full track endpoint URL directly instead of deriving it from api_host.
consumer	ConsumerLike	ThreadConsumer(Consumer(...))	Swap in a different delivery strategy — see Consumers.
request_timeout	float	90	HTTP request timeout in seconds.
retry_limit	int	4	Retry attempts on transient failures.
retry_backoff_factor	float	0.25	Exponential backoff base (seconds) between retries.
routes	dict	—	Override individual endpoint paths.

---

Core API

track(distinct_id, event_name, properties=None, meta=None, context=None)

Send a single custom event.

mf.track("user_123", "workspace_created", {"workspace_name": "Acme Inc"})

track_batch(events, context=None)

Send multiple events in one call — each item is a dict with event, and one of distinct_id/user_id/anonymous_id, plus optional properties.

mf.track_batch([
    {"event": "signup_started", "distinct_id": "user_123"},
    {"event": "signup_completed", "distinct_id": "user_123"},
])

identify(user_id, anonymous_id=None, traits=None, context=None)

Link a known user to their profile traits.

mf.identify("user_123", traits={"name": "Rahul Mehta", "plan": "pro"})

alias(alias_id, original, context=None)

Merge two identities (e.g. anonymous ID → logged-in user ID).

mf.alias("user_123", "anon_abc")

flush(timeout=None)

Force delivery of any buffered/queued events. Returns a list of consumer results (empty for consumers with no explicit flush, e.g. plain Consumer).

mf.flush(timeout=15.0)

---

User Profiles (people_*)

Set and update persistent user profile properties (Mixpanel-style profile operations).

mf.people_set("user_123", {"plan": "pro", "company": "Acme Inc"})
mf.people_set_once("user_123", {"first_seen": "2026-01-01T00:00:00Z"})
mf.people_increment("user_123", {"login_count": 1})
mf.people_append("user_123", {"tags": "beta_user"})
mf.people_union("user_123", {"roles": ["admin"]})
mf.people_unset("user_123", ["temp_flag"])

# Revenue-style running totals on the profile
mf.people_track_charge("user_123", 49, {"plan": "pro"})
mf.people_clear_charges("user_123")

Method	Description
people_set(distinct_id, properties)	Overwrite profile properties.
people_set_once(distinct_id, properties)	Set only if the property doesn't already exist.
people_increment(distinct_id, properties)	Increment numeric properties.
people_append(distinct_id, properties)	Append a value to a list property.
people_union(distinct_id, properties)	Add values to a list property without duplicates.
people_unset(distinct_id, properties)	Remove properties from the profile (list of names).
people_track_charge(distinct_id, amount, properties=None)	Append a transaction record.
people_clear_charges(distinct_id)	Clear all transaction records.

---

Consumers

The consumer you pass (or the default) controls how events are delivered:

Consumer	Behavior
Consumer	Synchronous — track()/identify() block and return the raw server response (or raise MetricFlowHttpError). Use this when you need a definitive pass/fail per call.
ThreadConsumer (default)	Wraps a Consumer and sends events on a background thread. Delivery failures are logged via the metricflow_ai logger, not raised — track() returns immediately and won't tell you if a send later failed.
BufferedConsumer	Buffers events in memory (default max_size=50) and flushes in batches.

from metricflow_ai import MetricFlow, Consumer

# Synchronous — get the real response back from track()/identify()
mf = MetricFlow(
    client_id="app_123",
    client_secret="secret_123",
    consumer=Consumer("app_123", "secret_123"),
)

If you need to see delivery failures from the default ThreadConsumer, configure logging:

import logging
logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s")

---

Revenue Tracking

Revenue is server-authenticated only — always initialise with a client_secret. Revenue sent from a browser/web context is rejected by the backend, so a malicious client can't inflate your numbers.

There are two ways to record revenue:

1. Custom events — add a $revenue property

Mark any event as revenue by attaching a $revenue amount and a currency. This is the easiest path for in-app purchases with your own event names.

mf.track("user_123", "order_completed", {
    "$revenue": 24.99,         # the amount — negative = refund
    "currency": "USD",         # any 3-letter ISO code (USD, EUR, INR, …)
    "insert_id": "order_789",  # unique id → dedup (required)
    "plan": "pro",             # any extra business props are fine
})

# A refund — negative $revenue
mf.track("user_123", "order_refunded", {
    "$revenue": -24.99,
    "currency": "USD",
    "insert_id": "refund_789",
})

$revenue is the only $-prefixed property the backend accepts; all other $…/__… keys are rejected.

2. Named billing events — revenue.* helpers

For payment-provider webhooks (Stripe / Apple / Google) with canonical names, use the typed helpers. The amount comes from amount; refund_created is stored as negative revenue automatically.

mf.revenue.payment_succeeded(
    user_id="user_123",
    event_id="stripe_evt_123",  # → insert_id + idempotency_key
    amount=99,
    currency="USD",
    provider="stripe",
)

Helpers: payment_succeeded, invoice_paid, refund_created, subscription_created, subscription_updated, subscription_cancelled.

Required fields (both paths)

Field	Notes
identity	distinct_id / user_id — required
amount	$revenue (non-zero) or amount (> 0)
currency	3-letter ISO code, case-insensitive
dedup id	insert_id / idempotency_key (helpers derive it from event_id)

Missing any of these → the event is rejected and no revenue is recorded. Revenue is aggregated downstream on the computed revenue property.

---

Error Handling

All API calls (with a synchronous Consumer) raise a subclass of MetricFlowException:

from metricflow_ai import MetricFlow, Consumer, MetricFlowConfigError, MetricFlowHttpError

mf = MetricFlow(
    client_id="app_123",
    client_secret="secret_123",
    consumer=Consumer("app_123", "secret_123"),
)

try:
    mf.track("user_123", "event_name")
except MetricFlowHttpError as err:
    print("HTTP failure:", err.status_code, err.body)
except MetricFlowConfigError as err:
    print("Bad config:", err)

Error	Raised when
MetricFlowConfigError	Invalid configuration (missing client_id/client_secret).
MetricFlowHttpError	The ingestion request failed. Has .status_code and .body.
MetricFlowException	Base class for both — catch this to handle any SDK error generically.

With the default ThreadConsumer, these are logged, not raised — see Consumers if you need track()/identify() to surface failures directly.

---

License

MIT — see LICENSE.