feat: add PostHog telemetry for wrapper usage tracking

Add anonymous telemetry to track Python wrapper usage patterns,
matching the telemetry implementation in the main TypeScript project.

- Track `wrapper_used` event with method (global/npx/error)
- Share user ID with TypeScript version (~/.promptfoo/promptfoo.yaml)
- Use same PostHog endpoint (https://a.promptfoo.app)
- Respect PROMPTFOO_DISABLE_TELEMETRY=1 opt-out
- Non-blocking, error-swallowing design

Dependencies added: posthog, pyyaml

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: mldangelo

pyproject.toml

@@ -28,13 +28,17 @@ classifiers = [
     "Operating System :: OS Independent",
 ]
 requires-python = ">=3.9"
-dependencies = []
+dependencies = [
+    "posthog>=3.0.0",
+    "pyyaml>=6.0.0",
+]
 
 [project.optional-dependencies]
 dev = [
     "pytest>=8.4.0",
     "mypy>=1.16.0",
     "ruff>=0.12.0",
+    "types-pyyaml>=6.0.0",
 ]
 
 [project.scripts]

src/promptfoo/cli.py

@@ -11,6 +11,8 @@
 import sys
 from typing import NoReturn, Optional
 
+from .telemetry import record_wrapper_used
+
 _WRAPPER_ENV = "PROMPTFOO_PY_WRAPPER"
 _WINDOWS_SHELL_EXTENSIONS = (".bat", ".cmd")
 
@@ -184,16 +186,19 @@ def main() -> NoReturn:
     # Build command: try external promptfoo first, fall back to npx
     promptfoo_path = None if os.environ.get(_WRAPPER_ENV) else _find_external_promptfoo()
     if promptfoo_path:
+        record_wrapper_used("global")
         cmd = [promptfoo_path] + sys.argv[1:]
         env = os.environ.copy()
         env[_WRAPPER_ENV] = "1"
         result = _run_command(cmd, env=env)
     else:
         npx_path = shutil.which("npx")
         if npx_path:
+            record_wrapper_used("npx")
             cmd = [npx_path, "-y", "promptfoo@latest"] + sys.argv[1:]
             result = _run_command(cmd)
         else:
+            record_wrapper_used("error")
             print("ERROR: Neither promptfoo nor npx is available.", file=sys.stderr)
             print("Please install promptfoo: npm install -g promptfoo", file=sys.stderr)
             print("Or ensure Node.js is properly installed.", file=sys.stderr)

src/promptfoo/telemetry.py

@@ -0,0 +1,198 @@
+"""
+Telemetry module for the promptfoo Python wrapper.
+
+Sends anonymous usage analytics to PostHog to help improve promptfoo.
+Telemetry can be disabled by setting PROMPTFOO_DISABLE_TELEMETRY=1.
+"""
+
+import atexit
+import os
+import platform
+import sys
+import uuid
+from pathlib import Path
+from typing import Any, Optional
+
+import yaml
+from posthog import Posthog
+
+from . import __version__
+
+# PostHog configuration - same as the main promptfoo TypeScript project
+_POSTHOG_HOST = "https://a.promptfoo.app"
+_POSTHOG_KEY = "phc_E5n5uHnDo2eREJL1uqX1cIlbkoRby4yFWt3V94HqRRg"
+
+
+def _get_env_bool(name: str) -> bool:
+    """Check if an environment variable is set to a truthy value."""
+    value = os.environ.get(name, "").lower()
+    return value in ("1", "true", "yes", "on")
+
+
+def _is_ci() -> bool:
+    """Detect if running in a CI environment."""
+    ci_env_vars = [
+        "CI",
+        "CONTINUOUS_INTEGRATION",
+        "GITHUB_ACTIONS",
+        "GITLAB_CI",
+        "CIRCLECI",
+        "TRAVIS",
+        "JENKINS_URL",
+        "BUILDKITE",
+        "TEAMCITY_VERSION",
+        "TF_BUILD",  # Azure Pipelines
+    ]
+    return any(os.environ.get(var) for var in ci_env_vars)
+
+
+def _get_config_dir() -> Path:
+    """Get the promptfoo config directory path."""
+    return Path.home() / ".promptfoo"
+
+
+def _read_global_config() -> dict[str, Any]:
+    """Read the global promptfoo config from ~/.promptfoo/promptfoo.yaml."""
+    config_file = _get_config_dir() / "promptfoo.yaml"
+    if config_file.exists():
+        try:
+            with open(config_file) as f:
+                config = yaml.safe_load(f)
+                return config if isinstance(config, dict) else {}
+        except Exception:
+            return {}
+    return {}
+
+
+def _write_global_config(config: dict[str, Any]) -> None:
+    """Write the global promptfoo config to ~/.promptfoo/promptfoo.yaml."""
+    config_dir = _get_config_dir()
+    config_dir.mkdir(parents=True, exist_ok=True)
+    config_file = config_dir / "promptfoo.yaml"
+    try:
+        with open(config_file, "w") as f:
+            yaml.dump(config, f, default_flow_style=False)
+    except Exception:
+        pass  # Silently fail - telemetry should never break the CLI
+
+
+def _get_user_id() -> str:
+    """Get or create a unique user ID stored in the global config."""
+    config = _read_global_config()
+    user_id = config.get("id")
+
+    if not user_id:
+        user_id = str(uuid.uuid4())
+        config["id"] = user_id
+        _write_global_config(config)
+
+    return user_id
+
+
+def _get_user_email() -> Optional[str]:
+    """Get the user email from the global config if set."""
+    config = _read_global_config()
+    account = config.get("account", {})
+    return account.get("email") if isinstance(account, dict) else None
+
+
+class _Telemetry:
+    """Internal telemetry client for the promptfoo Python wrapper."""
+
+    def __init__(self) -> None:
+        self._client: Optional[Posthog] = None
+        self._user_id: Optional[str] = None
+        self._email: Optional[str] = None
+        self._initialized = False
+
+    @property
+    def _disabled(self) -> bool:
+        """Check if telemetry is disabled."""
+        return _get_env_bool("PROMPTFOO_DISABLE_TELEMETRY") or _get_env_bool("IS_TESTING")
+
+    def _ensure_initialized(self) -> None:
+        """Lazily initialize the telemetry client."""
+        if self._initialized:
+            return
+
+        self._initialized = True
+
+        if self._disabled:
+            return
+
+        try:
+            self._user_id = _get_user_id()
+            self._email = _get_user_email()
+            self._client = Posthog(
+                project_api_key=_POSTHOG_KEY,
+                host=_POSTHOG_HOST,
+            )
+        except Exception:
+            self._client = None  # Silently fail
+
+    def record(self, event_name: str, properties: Optional[dict[str, Any]] = None) -> None:
+        """Record a telemetry event."""
+        if self._disabled:
+            return
+
+        self._ensure_initialized()
+
+        if not self._client or not self._user_id:
+            return
+
+        try:
+            enriched_properties: dict[str, Any] = {
+                **(properties or {}),
+                "packageVersion": __version__,
+                "pythonVersion": platform.python_version(),
+                "platform": sys.platform,
+                "isRunningInCi": _is_ci(),
+                "source": "python-wrapper",
+            }
+
+            # Only set email if present
+            if self._email:
+                enriched_properties["$set"] = {"email": self._email}
+
+            self._client.capture(
+                event=event_name,
+                distinct_id=self._user_id,
+                properties=enriched_properties,
+            )
+        except Exception:
+            pass  # Silently fail - telemetry should never break the CLI
+
+    def shutdown(self) -> None:
+        """Shutdown the telemetry client and flush any pending events."""
+        if self._client:
+            try:
+                self._client.flush()
+                self._client.shutdown()
+            except Exception:
+                pass  # Silently fail
+            finally:
+                self._client = None
+
+
+# Global singleton instance
+_telemetry: Optional[_Telemetry] = None
+
+
+def _get_telemetry() -> _Telemetry:
+    """Get the global telemetry instance."""
+    global _telemetry
+    if _telemetry is None:
+        _telemetry = _Telemetry()
+        atexit.register(_telemetry.shutdown)
+    return _telemetry
+
+
+def record_wrapper_used(method: str) -> None:
+    """
+    Record that the Python wrapper was used.
+
+    Args:
+        method: The execution method used - "global" for global promptfoo install,
+                "npx" for npx fallback, or "error" if execution failed.
+    """
+    _get_telemetry().record("wrapper_used", {"method": method, "wrapperType": "python"})