feat(cli): validate API key upfront via ping and unify telemetry identity resolution (#4701)

cli/python/src/mem0_cli/app.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import contextlib
 import json as _json
 import os
 import stat as _stat_mod
@@ -12,7 +13,7 @@ import typer
 from rich.console import Console
 
 from mem0_cli import __version__
-from mem0_cli.branding import BRAND_COLOR, print_error
+from mem0_cli.branding import BRAND_COLOR, print_error, print_warning
 
 console = Console()
 err_console = Console(stderr=True)
@@ -55,6 +56,10 @@ event_app = typer.Typer(
 # entity_app and event_app registered after Memory commands to control panel ordering
 
 
+# ── Validated user identity (set by _get_backend_and_config) ──────────────
+
+_validated_user_email: str | None = None
+
 # ── Telemetry helper ─────────────────────────────────────────────────────
 
 
@@ -66,7 +71,7 @@ def _fire_telemetry(command_name: str, extra: dict | None = None) -> None:
         props = {"command": command_name}
         if extra:
             props.update(extra)
-        capture_event(f"cli.{command_name}", props)
+        capture_event(f"cli.{command_name}", props, pre_resolved_email=_validated_user_email)
     except Exception:
         pass
 
@@ -96,9 +101,16 @@ def _get_backend_and_config(
     api_key: str | None = None,
     base_url: str | None = None,
 ):
-    """Build and return the Platform backend plus the loaded config."""
+    """Build and return the Platform backend plus the loaded config.
+
+    Validates the API key upfront via ``/v1/ping/`` and caches the
+    resolved user email for telemetry.
+    """
+    global _validated_user_email
+
     from mem0_cli.backend import get_backend
-    from mem0_cli.config import load_config
+    from mem0_cli.backend.platform import AuthError
+    from mem0_cli.config import load_config, save_config
 
     config = load_config()
 
@@ -115,7 +127,29 @@ def _get_backend_and_config(
         )
         raise typer.Exit(1)
 
-    return get_backend(config), config
+    backend = get_backend(config)
+
+    # Validate the API key upfront with a fast timeout
+    try:
+        ping_data = backend.ping(timeout=5.0)
+        email = ping_data.get("user_email") if isinstance(ping_data, dict) else None
+        if email:
+            _validated_user_email = email
+            if config.platform.user_email != email:
+                config.platform.user_email = email
+                with contextlib.suppress(Exception):
+                    save_config(config)
+    except AuthError:
+        print_error(
+            err_console,
+            "Invalid or expired API key.",
+            hint="Run 'mem0 init' or set MEM0_API_KEY environment variable.",
+        )
+        raise typer.Exit(1) from None
+    except Exception:
+        print_warning(err_console, "Could not validate API key (network issue). Proceeding anyway.")
+
+    return backend, config
 
 
 def _get_backend(

cli/python/src/mem0_cli/backend/platform.py

@@ -288,8 +288,18 @@ class PlatformBackend(Backend):
             result = self._request("DELETE", f"/v2/entities/{entity_type}/{entity_id}/")
         return result
 
-    def ping(self) -> dict:
-        """Call the ping endpoint and return the raw response."""
+    def ping(self, timeout: float | None = None) -> dict:
+        """Call the ping endpoint and return the raw response.
+
+        When *timeout* is given it overrides the client-level timeout so that
+        validation pings can fail fast without blocking the user.
+        """
+        if timeout is not None:
+            resp = self._client.get("/v1/ping/", timeout=timeout)
+            if resp.status_code == 401:
+                raise AuthError("Authentication failed. Your API key may be invalid or expired.")
+            resp.raise_for_status()
+            return resp.json()
         return self._request("GET", "/v1/ping/")
 
     def status(

cli/python/src/mem0_cli/telemetry.py

@@ -45,8 +45,17 @@ def _get_distinct_id() -> str:
     return "anonymous-cli"
 
 
-def capture_event(event_name: str, properties: dict[str, Any] | None = None) -> None:
-    """Fire a PostHog event via a detached subprocess (non-blocking)."""
+def capture_event(
+    event_name: str,
+    properties: dict[str, Any] | None = None,
+    pre_resolved_email: str | None = None,
+) -> None:
+    """Fire a PostHog event via a detached subprocess (non-blocking).
+
+    When *pre_resolved_email* is provided (e.g. from an upfront ping
+    validation), it is used directly as the PostHog distinct ID and the
+    subprocess skips its own ``/v1/ping/`` call.
+    """
     if not _is_telemetry_enabled():
         return
 
@@ -56,7 +65,7 @@ def capture_event(event_name: str, properties: dict[str, Any] | None = None) ->
         from mem0_cli.state import is_agent_mode
 
         config = load_config()
-        distinct_id = _get_distinct_id()
+        distinct_id = pre_resolved_email or _get_distinct_id()
 
         payload = {
             "api_key": POSTHOG_API_KEY,