feat(cli): add PostHog telemetry and source tracking to Python & Node CLIs (#4699)

cli/python/src/mem0_cli/telemetry_sender.py

@@ -0,0 +1,86 @@
+"""Standalone telemetry sender — runs as a detached subprocess.
+
+Usage: python -m mem0_cli.telemetry_sender '<json context>'
+
+This module is spawned by telemetry.capture_event() and runs independently
+of the parent CLI process. It:
+
+1. Resolves the user's email via /v1/ping/ if not already cached
+2. Caches the email in ~/.mem0/config.json for future runs
+3. Sends the PostHog event
+
+All errors are silently swallowed — this process must never produce output
+or affect the user experience.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import urllib.request
+
+
+def main() -> None:
+    ctx = json.loads(sys.argv[1])
+    payload = ctx["payload"]
+
+    if ctx.get("needs_email") and ctx.get("mem0_api_key"):
+        _resolve_and_cache_email(ctx, payload)
+
+    _send_posthog_event(ctx["posthog_host"], payload)
+
+
+def _resolve_and_cache_email(ctx: dict, payload: dict) -> None:
+    """Call /v1/ping/ to get the user's email, update the payload, and cache it."""
+    try:
+        ping_url = ctx["mem0_base_url"].rstrip("/") + "/v1/ping/"
+        req = urllib.request.Request(
+            ping_url,
+            headers={
+                "Authorization": "Token " + ctx["mem0_api_key"],
+                "Content-Type": "application/json",
+            },
+        )
+        resp = urllib.request.urlopen(req, timeout=10)
+        data = json.loads(resp.read())
+        email = data.get("user_email")
+        if email:
+            payload["distinct_id"] = email
+            _cache_email(ctx.get("config_path"), email)
+    except Exception:
+        pass
+
+
+def _cache_email(config_path: str | None, email: str) -> None:
+    """Write user_email into the config file for future runs."""
+    if not config_path:
+        return
+    try:
+        with open(config_path) as f:
+            cfg = json.load(f)
+        cfg.setdefault("platform", {})["user_email"] = email
+        with open(config_path, "w") as f:
+            json.dump(cfg, f, indent=2)
+    except Exception:
+        pass
+
+
+def _send_posthog_event(posthog_host: str, payload: dict) -> None:
+    """POST the event to PostHog."""
+    try:
+        body = json.dumps(payload).encode()
+        req = urllib.request.Request(
+            posthog_host,
+            data=body,
+            headers={"Content-Type": "application/json"},
+        )
+        urllib.request.urlopen(req, timeout=10)
+    except Exception:
+        pass
+
+
+if __name__ == "__main__":
+    import contextlib
+
+    with contextlib.suppress(Exception):
+        main()