diff --git a/.sampo/changesets/metrics-python-alpha.md b/.sampo/changesets/metrics-python-alpha.md
new file mode 100644
index 00000000..3134b4f3
--- /dev/null
+++ b/.sampo/changesets/metrics-python-alpha.md
@@ -0,0 +1,16 @@
+---
+pypi/posthog: minor
+---
+
+Add the `posthog.metrics` API (`count`, `gauge`, `histogram`) — alpha.
+
+Backend services can now record metrics through the same statsd-style pre-aggregating client the browser SDK ships, with no OpenTelemetry setup:
+
+```python
+client = Posthog("<ph_project_api_key>", metrics={"service_name": "billing-worker"})
+client.metrics.count("invoices.processed", 1, attributes={"plan": "pro"})
+client.metrics.gauge("queue.depth", 42)
+client.metrics.histogram("job.duration", 187, unit="ms")
+```
+
+Samples aggregate in memory and flush as OTLP/JSON to `/i/v1/metrics` (one data point per series per window, delta temporality). Pending metrics are flushed on `shutdown()`; buffered windows are retried on transient failures and dropped loudly after 3 consecutive failed flushes. The `metrics` client option accepts `service_name`, `service_version`, `environment`, `resource_attributes`, `flush_interval` (seconds), `max_series_per_flush` (cardinality guardrail, default 1000), and a `before_send` hook.
diff --git a/posthog/__init__.py b/posthog/__init__.py
index 9b664424..1842d467 100644
--- a/posthog/__init__.py
+++ b/posthog/__init__.py
@@ -1132,8 +1132,7 @@ def shutdown() -> None:
     Category:
         Client management
     """
-    _proxy("flush")
-    _proxy("join")
+    _proxy("shutdown")
 
 
 def setup() -> Client:
diff --git a/posthog/client.py b/posthog/client.py
index 7bcfb591..2458af21 100644
--- a/posthog/client.py
+++ b/posthog/client.py
@@ -16,6 +16,7 @@
 
 from posthog._async_utils import _BackgroundEventLoopRunner
 from posthog.args import ID_TYPES, ExceptionArg, OptionalCaptureArgs, OptionalSetArgs
+from posthog.metrics_capture import PostHogMetrics
 from posthog.capture_compression import (
     CaptureCompression,
     _resolve_capture_compression,
@@ -270,6 +271,7 @@ def __init__(
         capture_compression: Optional[Union[CaptureCompression, str]] = None,
         secret_key=None,
         _dedicated_ai_endpoint=False,
+        metrics: Optional[dict] = None,
     ):
         """
         Initialize a new PostHog client instance.
@@ -423,6 +425,9 @@ def __init__(
         self._flag_definition_cache_provider_async_runner_lock = threading.Lock()
         self.disabled = disabled or not self.api_key
         self.disable_geoip = disable_geoip
+        self._metrics_config = metrics
+        self._metrics: Optional[PostHogMetrics] = None
+        self._metrics_lock = threading.Lock()
         self.is_server = is_server
         self.historical_migration = historical_migration
         # Selects the capture wire protocol (V0 legacy `/batch/` vs V1
@@ -1599,6 +1604,12 @@ def _reinit_after_fork(self):
         self._flag_definition_cache_provider_async_runner = None
         self._flag_definition_cache_provider_async_runner_lock = threading.Lock()
 
+        # Metrics locks may have been held by a parent thread at fork time; replace
+        # them (never acquire them) so the child can't deadlock on a vanished holder.
+        self._metrics_lock = threading.Lock()
+        if self._metrics is not None:
+            self._metrics._reinit_after_fork()
+
         # If using Redis cache, we must reinitialize to get a fresh connection (fork-safe).
         # If using Memory cache, we keep it as-is to benefit from the inherited warm cache.
         if isinstance(self.flag_cache, RedisFlagCache):
@@ -1715,6 +1726,30 @@ def _enqueue(self, msg, disable_geoip):
             self.log.warning("analytics-python queue is full")
             return None
 
+    @property
+    def metrics(self) -> PostHogMetrics:
+        """
+        The `posthog.metrics` API: a statsd-style pre-aggregating metrics client — alpha.
+
+        Samples fold into per-series aggregates in memory and flush as one OTLP
+        data point per series per window, so recording from hot paths is cheap.
+        Configure via the ``metrics`` client option; pending metrics flush on
+        ``shutdown()``.
+
+        Examples:
+            ```python
+            client = Posthog("<ph_project_api_key>", metrics={"service_name": "billing-worker"})
+            client.metrics.count("invoices.processed", 1, attributes={"plan": "pro"})
+            client.metrics.gauge("queue.depth", 42)
+            client.metrics.histogram("job.duration", 187, unit="ms")
+            ```
+        """
+        if self._metrics is None:
+            with self._metrics_lock:
+                if self._metrics is None:
+                    self._metrics = PostHogMetrics(self, self._metrics_config)
+        return self._metrics
+
     def flush(self, timeout_seconds: Optional[float] = 10) -> None:
         """
         Force a flush from the internal queue to the server. Do not use directly, call `shutdown()` instead.
@@ -1789,6 +1824,12 @@ def shutdown(self) -> None:
             ```
         """
         self.flush(timeout_seconds=None)
+        if self._metrics is not None:
+            try:
+                self._metrics.flush()
+            except Exception:
+                self.log.exception("Failed to flush metrics on shutdown")
+            self._metrics.reset()
         self.join()
         self.distinct_ids_feature_flags_reported.clear()
 
diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py
new file mode 100644
index 00000000..a1bcf016
--- /dev/null
+++ b/posthog/metrics_capture.py
@@ -0,0 +1,661 @@
+"""Statsd-style pre-aggregating metrics client (`client.metrics`) — alpha.
+
+Samples fold into per-series aggregates in memory (counts sum, gauges keep the
+last value, histograms accumulate buckets) and flush as one OTLP/JSON data
+point per series per window to ``/i/v1/metrics`` — a burst of 10k ``count()``
+calls costs one data point on the wire. Sums and histograms use delta
+temporality, so each data point stands alone and process restarts need no
+cross-window state. Mirrors the ``posthog-js`` core implementation so every
+SDK speaks the same wire shape.
+
+Deliberately unlike event capture, no per-user context (distinct ID, session)
+is attached: every attribute value creates a new series, and per-user series
+are the canonical metrics-cardinality explosion.
+
+Delivery is at-least-once: a request that succeeds server-side but fails
+client-side (e.g. a read timeout) is retried with the next window, which can
+double-count that window's deltas. Windows are dropped loudly after
+``_MAX_CONSECUTIVE_SEND_FAILURES`` failed flushes.
+"""
+
+import gzip
+import json
+import logging
+import math
+import os
+import threading
+import time
+from typing import Any, Callable, Optional, Union
+from urllib.parse import quote
+
+import requests
+
+from posthog.request import _get_session
+from posthog.utils import remove_trailing_slash
+from posthog.version import VERSION
+
+log = logging.getLogger("posthog")
+
+MetricAttributeValue = Union[str, int, float, bool]
+
+# OpenTelemetry SDK default bucket boundaries — usable resolution for common
+# latency/size ranges without per-metric configuration. Must match posthog-js.
+DEFAULT_HISTOGRAM_BOUNDS = [
+    0,
+    5,
+    10,
+    25,
+    50,
+    75,
+    100,
+    250,
+    500,
+    750,
+    1000,
+    2500,
+    5000,
+    7500,
+    10000,
+]
+
+_OTLP_TEMPORALITY_DELTA = 1
+_VALID_METRIC_TYPES = ("count", "gauge", "histogram")
+# Consecutive failed flushes before the buffered window is dropped (loudly) — bounds
+# memory and payload growth against a permanently unreachable endpoint.
+_MAX_CONSECUTIVE_SEND_FAILURES = 3
+_DEFAULT_FLUSH_INTERVAL_SECONDS = 10.0
+_DEFAULT_MAX_SERIES_PER_FLUSH = 1000
+_SCOPE_NAME = "posthog-python"
+
+
+def _to_otlp_any_value(value: Any) -> dict:
+    # bool before int: Python bool is an int subclass and must not encode as intValue.
+    if isinstance(value, bool):
+        return {"boolValue": value}
+    if isinstance(value, int):
+        return {"intValue": value}
+    if isinstance(value, float):
+        # proto3 JSON has no representation for non-finite floats; encode the proto3
+        # literal strings (not Python's "inf"/"nan") so both SDKs emit identical bytes.
+        if not math.isfinite(value):
+            if math.isnan(value):
+                return {"stringValue": "NaN"}
+            return {"stringValue": "Infinity" if value > 0 else "-Infinity"}
+        # Integral floats encode as intValue, matching the JS Number.isInteger branch.
+        if value.is_integer():
+            return {"intValue": int(value)}
+        return {"doubleValue": value}
+    if isinstance(value, str):
+        return {"stringValue": value}
+    if isinstance(value, (list, tuple)):
+        return {"arrayValue": {"values": [_to_otlp_any_value(v) for v in value]}}
+    try:
+        return {"stringValue": json.dumps(value)}
+    except (TypeError, ValueError):
+        return {"stringValue": str(value)}
+
+
+def _to_otlp_key_value_list(attributes: dict) -> list:
+    # str(key): OTLP KeyValue.key is a string field — strict decoders reject numeric
+    # keys — and the series identity already stringifies keys the same way.
+    return [
+        {"key": str(key), "value": _to_otlp_any_value(value)}
+        for key, value in attributes.items()
+        if value is not None
+    ]
+
+
+def _ms_to_unix_nano(ms: int) -> str:
+    # OTLP requires nanoseconds as a decimal string (uint64).
+    return f"{ms}000000"
+
+
+def _bucket_index_for(value: float, bounds: list) -> int:
+    for i, bound in enumerate(bounds):
+        if value <= bound:
+            return i
+    return len(bounds)
+
+
+def _series_key(
+    metric_type: str, name: str, unit: Optional[str], attributes: Optional[dict]
+) -> str:
+    """Canonical, total series identity: JSON-encoded like the JS core's seriesKey, so any
+    attribute value the encoder accepts (lists, dicts, mixed keys) produces a hashable key,
+    and bool/int values stay distinct (json encodes true vs 1)."""
+    attrs_part = ""
+    if attributes:
+        items = sorted(
+            ((str(k), v) for k, v in attributes.items()), key=lambda kv: kv[0]
+        )
+        attrs_part = ",".join(
+            f"{json.dumps(k)}:{json.dumps(v, sort_keys=True, default=str)}"
+            for k, v in items
+        )
+    return "\x00".join((metric_type, name, unit or "", attrs_part))
+
+
+class _SeriesState:
+    __slots__ = (
+        "name",
+        "type",
+        "unit",
+        "attributes",
+        "window_start_ms",
+        "total",
+        "last",
+        "hist",
+    )
+
+    def __init__(
+        self,
+        name: str,
+        metric_type: str,
+        unit: Optional[str],
+        attributes: Optional[dict],
+    ):
+        self.name = name
+        self.type = metric_type
+        self.unit = unit
+        # Snapshot: the series key was computed from these values, so a caller
+        # mutating the dict after capture must not change the stored series.
+        self.attributes = dict(attributes) if attributes else None
+        self.window_start_ms = int(time.time() * 1000)
+        self.total: Optional[float] = None
+        self.last: Optional[float] = None
+        self.hist: Optional[dict] = None
+
+
+class PostHogMetrics:
+    """The ``client.metrics`` API: ``count``, ``gauge``, ``histogram``, ``flush``.
+
+    Thread-safe; safe to call from hot paths. Configure via the ``metrics``
+    client option (``flush_interval`` is in seconds, matching the client's own
+    ``flush_interval`` — unlike posthog-js, whose ``flushIntervalMs`` is milliseconds)::
+
+        client = Client("phc_...", metrics={"service_name": "billing-worker"})
+        client.metrics.count("invoices.processed", 1, attributes={"plan": "pro"})
+        client.metrics.gauge("queue.depth", 42)
+        client.metrics.histogram("job.duration", 187, unit="ms")
+    """
+
+    def __init__(self, client, config: Optional[dict] = None):
+        self._client = client
+        config = config or {}
+        resource_attributes = config.get("resource_attributes") or {}
+        self._service_name: Optional[str] = resource_attributes.get(
+            "service.name"
+        ) or config.get("service_name")
+        self._service_version: Optional[str] = resource_attributes.get(
+            "service.version"
+        ) or config.get("service_version")
+        self._environment: Optional[str] = resource_attributes.get(
+            "deployment.environment"
+        ) or config.get("environment")
+        self._resource_attributes: dict = resource_attributes
+        self._flush_interval: float = config.get(
+            "flush_interval", _DEFAULT_FLUSH_INTERVAL_SECONDS
+        )
+        self._max_series_per_flush: int = config.get(
+            "max_series_per_flush", _DEFAULT_MAX_SERIES_PER_FLUSH
+        )
+        self._before_send: Optional[Callable] = config.get("before_send")
+
+        self._lock = threading.Lock()
+        self._pid = os.getpid()
+        self._consecutive_send_failures = 0
+        self._capture_error_warned = False
+        # Serializes flushes so a manual flush() can't race a timer flush for the same window.
+        self._flush_lock = threading.Lock()
+        self._series: dict = {}
+        self._flush_timer: Optional[threading.Timer] = None
+        self._series_cap_warned = False
+        self._type_by_name: dict = {}
+        self._type_collision_warned: set = set()
+
+    def count(
+        self,
+        name: str,
+        value: float = 1,
+        unit: Optional[str] = None,
+        attributes: Optional[dict] = None,
+    ) -> None:
+        """Record an increment for a monotonic counter (things that only go up)."""
+        self._guarded_capture("count", name, value, unit, attributes)
+
+    def gauge(
+        self,
+        name: str,
+        value: float,
+        unit: Optional[str] = None,
+        attributes: Optional[dict] = None,
+    ) -> None:
+        """Record the current value of something that goes up and down."""
+        self._guarded_capture("gauge", name, value, unit, attributes)
+
+    def histogram(
+        self,
+        name: str,
+        value: float,
+        unit: Optional[str] = None,
+        attributes: Optional[dict] = None,
+    ) -> None:
+        """Record one observation of a distribution (durations, sizes)."""
+        self._guarded_capture("histogram", name, value, unit, attributes)
+
+    def flush(self) -> None:
+        """Sends everything aggregated so far without waiting for the flush interval."""
+        with self._flush_lock:
+            self._do_flush()
+
+    def reset(self) -> None:
+        """Clears the flush timer and drops the current window."""
+        with self._lock:
+            self._clear_flush_timer()
+            self._series = {}
+            self._series_cap_warned = False
+            self._type_by_name = {}
+            self._type_collision_warned = set()
+
+    def _guarded_capture(
+        self,
+        metric_type: str,
+        name: str,
+        value: float,
+        unit: Optional[str],
+        attributes: Optional[dict],
+    ) -> None:
+        # A telemetry call must never raise into the host application, whatever the input.
+        try:
+            self._capture(metric_type, name, value, unit, attributes)
+        except Exception as e:
+            if not self._capture_error_warned:
+                self._capture_error_warned = True
+                log.warning("Dropping metric '%s': %s", name, e)
+
+    def _capture(
+        self,
+        metric_type: str,
+        name: str,
+        value: float,
+        unit: Optional[str],
+        attributes: Optional[dict],
+    ) -> None:
+        if getattr(self._client, "disabled", False):
+            return
+
+        sample = {
+            "name": name,
+            "type": metric_type,
+            "value": value,
+            "unit": unit,
+            "attributes": attributes,
+        }
+        if self._before_send is not None:
+            try:
+                filtered = self._before_send(sample)
+            except Exception as e:
+                log.error("Error in metrics before_send: %s", e)
+                return
+            if not filtered:
+                return
+            if not isinstance(filtered, dict):
+                log.warning(
+                    "Dropping metric: before_send must return the sample dict or a falsy value"
+                )
+                return
+            sample_dict: dict[str, Any] = filtered
+            # Defaults keep the static types closed; a hook that removed the field
+            # produces a value the validation below drops.
+            name = sample_dict.get("name", "")
+            metric_type = sample_dict.get("type", metric_type)
+            value = sample_dict.get("value", math.nan)
+            unit = sample_dict.get("unit")
+            attributes = sample_dict.get("attributes")
+
+        if metric_type not in _VALID_METRIC_TYPES:
+            log.warning(
+                "Dropping metric '%s': unknown metric type '%s'", name, metric_type
+            )
+            return
+
+        if not name or not isinstance(name, str):
+            log.warning("Dropping metric with empty name")
+            return
+        if (
+            not isinstance(value, (int, float))
+            or isinstance(value, bool)
+            or not math.isfinite(value)
+        ):
+            log.warning("Dropping metric '%s': value must be a finite number", name)
+            return
+        if metric_type == "count" and value < 0:
+            log.warning(
+                "Dropping count '%s': counters are monotonic, value must be >= 0", name
+            )
+            return
+
+        if attributes:
+            # None-valued attributes are stripped from the wire, so strip them from the
+            # series identity too — otherwise two indistinguishable data points emit.
+            attributes = {k: v for k, v in attributes.items() if v is not None}
+        key = _series_key(metric_type, name, unit, attributes)
+
+        with self._lock:
+            self._reset_after_fork_locked()
+
+            state = self._series.get(key)
+            if state is None:
+                if len(self._series) >= self._max_series_per_flush:
+                    if not self._series_cap_warned:
+                        self._series_cap_warned = True
+                        log.warning(
+                            "Metric series cap reached (%s per flush window); dropping new series "
+                            "until the next flush. Reduce attribute cardinality.",
+                            self._max_series_per_flush,
+                        )
+                    return
+                state = _SeriesState(name, metric_type, unit, attributes)
+                self._series[key] = state
+
+            # Bookkeeping only for admitted samples, so name-cardinality misuse (IDs in
+            # metric names) can't grow this map past the series cap.
+            seen_type = self._type_by_name.get(name)
+            if seen_type is None:
+                self._type_by_name[name] = metric_type
+            elif seen_type != metric_type and name not in self._type_collision_warned:
+                self._type_collision_warned.add(name)
+                log.warning(
+                    "Metric name '%s' is already used as a %s; recording it as a %s too will blend "
+                    "both series in charts. Use a distinct name.",
+                    name,
+                    seen_type,
+                    metric_type,
+                )
+
+            self._fold(state, float(value))
+            self._arm_flush_timer()
+
+    def _reinit_after_fork(self) -> None:
+        # Runs in a forked child (via the client's os.register_at_fork hook) before
+        # user code. The inherited locks may be held by parent threads that do not
+        # exist in the child, so replace them without ever acquiring them.
+        self._lock = threading.Lock()
+        self._flush_lock = threading.Lock()
+        self._pid = os.getpid()
+        self._drop_inherited_window()
+
+    def _reset_after_fork_locked(self) -> None:
+        # PID-guard fallback for platforms without os.register_at_fork: a forked child
+        # inherits the parent's window and a timer handle whose thread does not exist
+        # in the child — without this, the child never flushes (silent total loss) and
+        # would duplicate the parent's samples if it ever did. Drop both.
+        pid = os.getpid()
+        if pid == self._pid:
+            return
+        self._pid = pid
+        self._drop_inherited_window()
+
+    def _drop_inherited_window(self) -> None:
+        self._flush_timer = None
+        self._series = {}
+        self._series_cap_warned = False
+        self._type_by_name = {}
+        self._type_collision_warned = set()
+        self._consecutive_send_failures = 0
+
+    def _fold(self, state: _SeriesState, value: float) -> None:
+        if state.type == "count":
+            state.total = (state.total or 0.0) + value
+        elif state.type == "gauge":
+            state.last = value
+        else:
+            hist = state.hist
+            if hist is None:
+                hist = state.hist = {
+                    "count": 0,
+                    "sum": 0.0,
+                    "min": value,
+                    "max": value,
+                    "bucket_counts": [0] * (len(DEFAULT_HISTOGRAM_BOUNDS) + 1),
+                }
+            hist["count"] += 1
+            hist["sum"] += value
+            hist["min"] = min(hist["min"], value)
+            hist["max"] = max(hist["max"], value)
+            hist["bucket_counts"][
+                _bucket_index_for(value, DEFAULT_HISTOGRAM_BOUNDS)
+            ] += 1
+
+    def _arm_flush_timer(self) -> None:
+        if self._flush_timer is not None:
+            return
+        timer = threading.Timer(self._flush_interval, self._timer_flush)
+        timer.daemon = True
+        self._flush_timer = timer
+        timer.start()
+
+    def _timer_flush(self) -> None:
+        with self._lock:
+            self._flush_timer = None
+        try:
+            self.flush()
+        except Exception as e:
+            log.error("Metrics flush failed: %s", e)
+
+    def _clear_flush_timer(self) -> None:
+        if self._flush_timer is not None:
+            self._flush_timer.cancel()
+            self._flush_timer = None
+
+    def _do_flush(self) -> None:
+        # Snapshot and reset the window under the lock; send outside it so
+        # captures during the request fold into a fresh window.
+        with self._lock:
+            if not self._series:
+                return
+            window = self._series
+            self._series = {}
+            self._series_cap_warned = False
+            self._type_by_name = {}
+            self._type_collision_warned = set()
+
+        # send=False mirrors event capture: recording succeeds locally, but
+        # nothing is transmitted — the flushed window is discarded.
+        if not getattr(self._client, "send", True):
+            return
+
+        payload = self._build_payload(window)
+        outcome = self._send(payload)
+        if outcome == "retry-later":
+            with self._lock:
+                self._consecutive_send_failures += 1
+                if self._consecutive_send_failures > _MAX_CONSECUTIVE_SEND_FAILURES:
+                    # A persistently unreachable endpoint must not buffer forever: drop the
+                    # window loudly instead of growing until a too-large drop loses more.
+                    log.error(
+                        "Dropping %s metric series after %s consecutive failed flushes — "
+                        "check the endpoint and network configuration",
+                        len(window),
+                        self._consecutive_send_failures,
+                    )
+                    self._consecutive_send_failures = 0
+                    return
+                # Transient failure: merge the unsent window back so the data rides the
+                # next flush instead of being lost — and re-arm the timer, since with no
+                # new captures nothing else would schedule that flush.
+                log.warning(
+                    "Metrics flush failed (attempt %s of %s); will retry with the next window",
+                    self._consecutive_send_failures,
+                    _MAX_CONSECUTIVE_SEND_FAILURES + 1,
+                )
+                self._merge_window_back(window)
+                self._arm_flush_timer()
+        elif outcome == "too-large":
+            log.warning(
+                "Metrics batch exceeded the server size limit and was dropped. "
+                "Reduce series count or attribute cardinality."
+            )
+            with self._lock:
+                self._consecutive_send_failures = 0
+        else:
+            with self._lock:
+                self._consecutive_send_failures = 0
+
+    def _send(self, payload: dict) -> str:
+        url = "{}/i/v1/metrics?token={}".format(
+            remove_trailing_slash(self._client.host),
+            quote(self._client.api_key, safe=""),
+        )
+        body = gzip.compress(json.dumps(payload).encode("utf-8"))
+        timeout = getattr(self._client, "timeout", 15) or 15
+        try:
+            # The shared pooled session: keepalive between the 10s flushes, fork-safe
+            # reset, and the same adapter/proxy configuration as event capture.
+            response = _get_session().post(
+                url,
+                data=body,
+                headers={
+                    "Content-Type": "application/json",
+                    "Content-Encoding": "gzip",
+                },
+                timeout=timeout,
+            )
+        except requests.exceptions.RequestException:
+            return "retry-later"
+        if response.status_code < 300:
+            return "ok"
+        if response.status_code == 413:
+            return "too-large"
+        if response.status_code >= 500 or response.status_code == 429:
+            return "retry-later"
+        log.error("Failed to send metrics batch: HTTP %s", response.status_code)
+        return "fatal"
+
+    def _merge_window_back(self, window: dict) -> None:
+        dropped = 0
+        for key, old in window.items():
+            current = self._series.get(key)
+            if current is None:
+                # The cap applies through merge-back too, or a long outage with attribute
+                # churn grows the live window (and the retried payload) without bound.
+                if len(self._series) >= self._max_series_per_flush:
+                    dropped += 1
+                    continue
+                self._series[key] = old
+                continue
+            current.window_start_ms = min(current.window_start_ms, old.window_start_ms)
+            if current.type == "count":
+                current.total = (current.total or 0.0) + (old.total or 0.0)
+            elif current.type == "histogram" and old.hist:
+                if current.hist is None:
+                    current.hist = old.hist
+                else:
+                    current.hist["count"] += old.hist["count"]
+                    current.hist["sum"] += old.hist["sum"]
+                    current.hist["min"] = min(current.hist["min"], old.hist["min"])
+                    current.hist["max"] = max(current.hist["max"], old.hist["max"])
+                    for i, count in enumerate(old.hist["bucket_counts"]):
+                        current.hist["bucket_counts"][i] += count
+            # Gauge: the live window's value is newer — keep it.
+        if dropped:
+            log.warning(
+                "Dropped %s unsent metric series while merging a failed flush back (series cap %s)",
+                dropped,
+                self._max_series_per_flush,
+            )
+
+    def _build_payload(self, window: dict) -> dict:
+        # User resource attributes first, SDK-controlled keys layered on top so
+        # a stray user key can't clobber attribution.
+        resource_attributes = dict(self._resource_attributes)
+        resource_attributes["service.name"] = self._service_name or "unknown_service"
+        if self._environment:
+            resource_attributes["deployment.environment"] = self._environment
+        if self._service_version:
+            resource_attributes["service.version"] = self._service_version
+        resource_attributes["telemetry.sdk.name"] = _SCOPE_NAME
+        resource_attributes["telemetry.sdk.version"] = VERSION
+
+        return {
+            "resourceMetrics": [
+                {
+                    "resource": {
+                        "attributes": _to_otlp_key_value_list(resource_attributes)
+                    },
+                    "scopeMetrics": [
+                        {
+                            "scope": {"name": _SCOPE_NAME, "version": VERSION},
+                            "metrics": self._build_metrics(window),
+                        }
+                    ],
+                }
+            ]
+        }
+
+    def _build_metrics(self, window: dict) -> list:
+        # One OTLP metric entry per (type, name, unit), one data point per attribute set.
+        now_nano = _ms_to_unix_nano(int(time.time() * 1000))
+        by_metric: dict = {}
+
+        for state in window.values():
+            metric_key = (state.type, state.name, state.unit or "")
+            metric = by_metric.get(metric_key)
+            if metric is None:
+                metric = {"name": state.name}
+                if state.unit:
+                    metric["unit"] = state.unit
+                if state.type == "count":
+                    metric["sum"] = {
+                        "aggregationTemporality": _OTLP_TEMPORALITY_DELTA,
+                        "isMonotonic": True,
+                        "dataPoints": [],
+                    }
+                elif state.type == "gauge":
+                    metric["gauge"] = {"dataPoints": []}
+                else:
+                    metric["histogram"] = {
+                        "aggregationTemporality": _OTLP_TEMPORALITY_DELTA,
+                        "dataPoints": [],
+                    }
+                by_metric[metric_key] = metric
+
+            attributes = _to_otlp_key_value_list(state.attributes or {})
+            start_nano = _ms_to_unix_nano(state.window_start_ms)
+
+            if state.type == "count":
+                metric["sum"]["dataPoints"].append(
+                    {
+                        "attributes": attributes,
+                        "startTimeUnixNano": start_nano,
+                        "timeUnixNano": now_nano,
+                        "asDouble": state.total or 0.0,
+                    }
+                )
+            elif state.type == "gauge":
+                metric["gauge"]["dataPoints"].append(
+                    {
+                        "attributes": attributes,
+                        "timeUnixNano": now_nano,
+                        "asDouble": state.last or 0.0,
+                    }
+                )
+            elif state.hist is not None:
+                # Encoding pinned by the ingest's JSON deserializer: nano timestamps are decimal
+                # strings, but count/bucketCounts are plain JSON numbers — string-encoded u64s in
+                # those fields are silently dropped upstream (opentelemetry-rust#3328).
+                metric["histogram"]["dataPoints"].append(
+                    {
+                        "attributes": attributes,
+                        "startTimeUnixNano": start_nano,
+                        "timeUnixNano": now_nano,
+                        "count": state.hist["count"],
+                        "sum": state.hist["sum"],
+                        "min": state.hist["min"],
+                        "max": state.hist["max"],
+                        "bucketCounts": state.hist["bucket_counts"],
+                        "explicitBounds": DEFAULT_HISTOGRAM_BOUNDS,
+                    }
+                )
+
+        return list(by_metric.values())
diff --git a/posthog/test/test_client_fork.py b/posthog/test/test_client_fork.py
index f5ed4279..eabae53b 100644
--- a/posthog/test/test_client_fork.py
+++ b/posthog/test/test_client_fork.py
@@ -1,5 +1,6 @@
 import os
 import gc
+import signal
 import unittest
 import warnings
 import weakref
@@ -237,6 +238,46 @@ def child_probe():
         )
         self.assertEqual(result, "ok")
 
+    def test_register_at_fork_replaces_metrics_locks_in_child_process(self):
+        # Locks held at fork time are inherited locked, and their holders don't
+        # exist in the child — the metrics path must not deadlock on them.
+        client = Client(FAKE_TEST_API_KEY, send=False)
+        client.metrics.count("parent.metric")
+
+        locks = [
+            client._metrics_lock,
+            client.metrics._lock,
+            client.metrics._flush_lock,
+        ]
+
+        def child_probe():
+            # A deadlocked child gets killed by SIGALRM (a signaled exit fails
+            # the assertion below) instead of hanging the test forever.
+            signal.alarm(5)
+            try:
+                if not client._metrics_lock.acquire(blocking=False):
+                    return "inherited _metrics_lock still held"
+                client._metrics_lock.release()
+                client.metrics.count("child.metric")
+                client.metrics.flush()
+            finally:
+                signal.alarm(0)
+            return "ok"
+
+        for lock in locks:
+            lock.acquire()
+        try:
+            status, result = self._run_fork_probe(child_probe)
+        finally:
+            for lock in locks:
+                lock.release()
+            client.metrics.reset()
+
+        self.assertTrue(
+            os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, msg=result
+        )
+        self.assertEqual(result, "ok")
+
     def test_register_at_fork_reinitializes_poller_and_sessions_in_child_process(self):
         client = Client(
             FAKE_TEST_API_KEY,
diff --git a/posthog/test/test_metrics.py b/posthog/test/test_metrics.py
new file mode 100644
index 00000000..b6783204
--- /dev/null
+++ b/posthog/test/test_metrics.py
@@ -0,0 +1,485 @@
+import gzip
+import json
+import math
+from unittest import mock
+
+import pytest
+
+import posthog
+from posthog.client import Client
+from posthog.metrics_capture import DEFAULT_HISTOGRAM_BOUNDS
+from posthog.version import VERSION
+
+FAKE_API_KEY = "phc_test_key"
+
+
+@pytest.fixture
+def client():
+    c = Client(FAKE_API_KEY, host="https://us.example.com", sync_mode=True)
+    yield c
+    c.metrics.reset()
+
+
+def mock_session(status_code=200):
+    session = mock.Mock()
+    session.post.return_value = mock.Mock(status_code=status_code)
+    return session
+
+
+def sent_payload(session):
+    """Decode the gzipped OTLP/JSON body of the last send through the mocked session."""
+    args, kwargs = session.post.call_args
+    body = kwargs.get("data")
+    return json.loads(gzip.decompress(body).decode("utf-8"))
+
+
+def flush_and_capture(client):
+    """Flush metrics with the HTTP boundary mocked; returns (payload, url, kwargs) of the send."""
+    session = mock_session()
+    with mock.patch("posthog.metrics_capture._get_session", return_value=session):
+        client.metrics.flush()
+        if not session.post.called:
+            return None, None, None
+        args, kwargs = session.post.call_args
+        return sent_payload(session), args[0] if args else kwargs.get("url"), kwargs
+
+
+def metrics_from(payload):
+    return payload["resourceMetrics"][0]["scopeMetrics"][0]["metrics"]
+
+
+class TestMetricsAggregation:
+    def test_count_burst_folds_to_one_delta_data_point(self, client):
+        # The whole point of pre-aggregation: 1000 count() calls must produce ONE data point
+        # whose value is the sum, marked delta + monotonic so the ingest diffs nothing.
+        for _ in range(1000):
+            client.metrics.count("jobs.processed")
+        client.metrics.count("jobs.processed", 5)
+
+        payload, url, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        assert metric["name"] == "jobs.processed"
+        assert metric["sum"]["aggregationTemporality"] == 1
+        assert metric["sum"]["isMonotonic"] is True
+        (dp,) = metric["sum"]["dataPoints"]
+        assert dp["asDouble"] == 1005.0
+        assert url == "https://us.example.com/i/v1/metrics?token=phc_test_key"
+
+    def test_gauge_keeps_last_value(self, client):
+        client.metrics.gauge("queue.depth", 10)
+        client.metrics.gauge("queue.depth", 3)
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        (dp,) = metric["gauge"]["dataPoints"]
+        assert dp["asDouble"] == 3.0
+        # Gauges are instantaneous: no window start on the data point (matches posthog-js).
+        assert "startTimeUnixNano" not in dp
+
+    def test_histogram_wire_shape(self, client):
+        # Pins the exact OTLP/JSON encoding the ingest's deserializer requires: nano timestamps
+        # as strings, but count/bucketCounts as plain JSON numbers (string-encoded u64s are
+        # silently dropped upstream — opentelemetry-rust#3328).
+        client.metrics.histogram("job.duration", 3, unit="ms")
+        client.metrics.histogram("job.duration", 40, unit="ms")
+        client.metrics.histogram("job.duration", 99999, unit="ms")
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        assert metric["unit"] == "ms"
+        assert metric["histogram"]["aggregationTemporality"] == 1
+        (dp,) = metric["histogram"]["dataPoints"]
+        assert dp["count"] == 3
+        assert dp["sum"] == 100042.0
+        assert dp["min"] == 3.0
+        assert dp["max"] == 99999.0
+        assert dp["explicitBounds"] == DEFAULT_HISTOGRAM_BOUNDS
+        assert isinstance(dp["count"], int)
+        assert all(isinstance(c, int) for c in dp["bucketCounts"])
+        assert len(dp["bucketCounts"]) == len(DEFAULT_HISTOGRAM_BOUNDS) + 1
+        assert sum(dp["bucketCounts"]) == 3
+        # 3 → first bucket with bound >= 3 (index 1: bound 5); 99999 → overflow bucket.
+        assert dp["bucketCounts"][1] == 1
+        assert dp["bucketCounts"][-1] == 1
+        assert isinstance(dp["timeUnixNano"], str)
+        assert isinstance(dp["startTimeUnixNano"], str)
+
+    def test_attribute_sets_split_series_and_order_does_not(self, client):
+        client.metrics.count(
+            "http.requests", 1, attributes={"route": "/a", "status": "200"}
+        )
+        client.metrics.count(
+            "http.requests", 1, attributes={"status": "200", "route": "/a"}
+        )
+        client.metrics.count(
+            "http.requests", 1, attributes={"route": "/b", "status": "200"}
+        )
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        points = metric["sum"]["dataPoints"]
+        assert len(points) == 2
+        values = sorted(dp["asDouble"] for dp in points)
+        assert values == [1.0, 2.0]
+
+    def test_attribute_value_otlp_encoding(self, client):
+        # bool must encode as boolValue, not intValue — Python bool is an int subclass, so a
+        # naive isinstance(int) check first silently miscodes True as 1.
+        client.metrics.count(
+            "encoded", 1, attributes={"s": "x", "b": True, "i": 7, "f": 1.5}
+        )
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        (dp,) = metric["sum"]["dataPoints"]
+        by_key = {attr["key"]: attr["value"] for attr in dp["attributes"]}
+        assert by_key["s"] == {"stringValue": "x"}
+        assert by_key["b"] == {"boolValue": True}
+        assert by_key["i"] == {"intValue": 7}
+        assert by_key["f"] == {"doubleValue": 1.5}
+
+
+class TestMetricsGuardrails:
+    def test_series_cap_drops_new_series_not_existing(self, client):
+        capped = Client(
+            FAKE_API_KEY,
+            host="https://us.example.com",
+            sync_mode=True,
+            metrics={"max_series_per_flush": 2},
+        )
+        capped.metrics.count("m", 1, attributes={"k": "a"})
+        capped.metrics.count("m", 1, attributes={"k": "b"})
+        capped.metrics.count(
+            "m", 1, attributes={"k": "c"}
+        )  # new series past cap: dropped
+        capped.metrics.count(
+            "m", 1, attributes={"k": "a"}
+        )  # existing series: still folds
+
+        payload, _, _ = flush_and_capture(capped)
+        capped.metrics.reset()
+
+        (metric,) = metrics_from(payload)
+        points = metric["sum"]["dataPoints"]
+        assert len(points) == 2
+        assert sorted(dp["asDouble"] for dp in points) == [1.0, 2.0]
+
+    @pytest.mark.parametrize(
+        "record",
+        [
+            lambda m: m.count("bad", -1),  # counters are monotonic
+            lambda m: m.count("bad", math.nan),
+            lambda m: m.gauge("bad", math.inf),
+            lambda m: m.count("", 1),  # empty name
+        ],
+    )
+    def test_invalid_samples_dropped(self, client, record):
+        record(client.metrics)
+
+        payload, _, _ = flush_and_capture(client)
+
+        assert payload is None  # nothing aggregated, nothing sent
+
+    def test_disabled_client_records_nothing(self):
+        disabled = Client(FAKE_API_KEY, host="https://us.example.com", disabled=True)
+        disabled.metrics.count("m", 1)
+
+        payload, _, _ = flush_and_capture(disabled)
+
+        assert payload is None
+
+
+class TestMetricsDelivery:
+    def test_resource_attributes_layer_sdk_keys_over_user_keys(self):
+        c = Client(
+            FAKE_API_KEY,
+            host="https://us.example.com",
+            sync_mode=True,
+            metrics={
+                "service_name": "billing-worker",
+                "environment": "production",
+                # A stray user key must not clobber SDK attribution.
+                "resource_attributes": {
+                    "telemetry.sdk.name": "spoofed",
+                    "team": "billing",
+                },
+            },
+        )
+        c.metrics.count("m", 1)
+
+        payload, _, _ = flush_and_capture(c)
+        c.metrics.reset()
+
+        attrs = {
+            attr["key"]: attr["value"]
+            for attr in payload["resourceMetrics"][0]["resource"]["attributes"]
+        }
+        assert attrs["service.name"] == {"stringValue": "billing-worker"}
+        assert attrs["deployment.environment"] == {"stringValue": "production"}
+        assert attrs["team"] == {"stringValue": "billing"}
+        assert attrs["telemetry.sdk.name"] == {"stringValue": "posthog-python"}
+        assert attrs["telemetry.sdk.version"] == {"stringValue": VERSION}
+        scope = payload["resourceMetrics"][0]["scopeMetrics"][0]["scope"]
+        assert scope == {"name": "posthog-python", "version": VERSION}
+
+    def test_transient_failure_merges_window_back(self, client):
+        # A 5xx must not lose the window: the counts ride the next flush, summed with new samples.
+        client.metrics.count("m", 3)
+        with mock.patch(
+            "posthog.metrics_capture._get_session", return_value=mock_session(503)
+        ):
+            client.metrics.flush()
+
+        client.metrics.count("m", 4)
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        (dp,) = metric["sum"]["dataPoints"]
+        assert dp["asDouble"] == 7.0
+
+    @pytest.mark.parametrize(
+        "flush_via",
+        [
+            lambda m: m.flush(),  # manual flush
+            lambda m: m._timer_flush(),  # what the flush timer invokes
+        ],
+        ids=["manual", "timer"],
+    )
+    def test_send_false_aggregates_but_never_posts(self, flush_via):
+        # send=False documents "queueing succeeds but events are not sent"; metrics
+        # must mirror that: fold locally, never hit the transport.
+        c = Client(FAKE_API_KEY, host="https://us.example.com", send=False, thread=0)
+        c.metrics.count("jobs.processed")
+
+        session = mock_session()
+        with mock.patch("posthog.metrics_capture._get_session", return_value=session):
+            flush_via(c.metrics)
+        c.metrics.reset()
+
+        assert not session.post.called
+
+    def test_shutdown_flushes_pending_metrics(self):
+        c = Client(FAKE_API_KEY, host="https://us.example.com", sync_mode=True)
+        c.metrics.count("m", 1)
+
+        session = mock_session()
+        with mock.patch("posthog.metrics_capture._get_session", return_value=session):
+            c.shutdown()
+
+        assert session.post.called
+        payload = sent_payload(session)
+        (metric,) = metrics_from(payload)
+        assert metric["name"] == "m"
+
+    def test_module_shutdown_flushes_default_client_metrics(self):
+        # Module-level shutdown() must delegate to Client.shutdown(), or the default
+        # client from posthog.setup() loses its final metrics window.
+        saved = (
+            posthog.default_client,
+            posthog.api_key,
+            posthog.host,
+            posthog.sync_mode,
+        )
+        posthog.default_client = None
+        posthog.api_key = FAKE_API_KEY
+        posthog.host = "https://us.example.com"
+        posthog.sync_mode = True
+        try:
+            posthog.setup().metrics.count("m", 1)
+
+            session = mock_session()
+            with mock.patch(
+                "posthog.metrics_capture._get_session", return_value=session
+            ):
+                posthog.shutdown()
+
+            assert session.post.called
+            (metric,) = metrics_from(sent_payload(session))
+            assert metric["name"] == "m"
+
+            # The window was flushed and cleared, not left pending.
+            payload, _, _ = flush_and_capture(posthog.default_client)
+            assert payload is None
+        finally:
+            (
+                posthog.default_client,
+                posthog.api_key,
+                posthog.host,
+                posthog.sync_mode,
+            ) = saved
+
+
+class TestMetricsCrashSafety:
+    # A telemetry SDK must never raise into the host application — these inputs all
+    # crashed the capture hot path before the series key became JSON-based and
+    # _capture gained validation (QA findings, reproduced).
+
+    @pytest.mark.parametrize(
+        "attributes",
+        [
+            {"tags": ["a", "b"]},  # unhashable value
+            {"nested": {"k": "v"}},  # unhashable value
+            {"a": 1, 2: "x"},  # unsortable mixed-type keys
+        ],
+    )
+    def test_hostile_attributes_do_not_raise(self, client, attributes):
+        client.metrics.count("hostile", 1, attributes=attributes)  # must not raise
+
+    def test_non_string_attribute_keys_stringify_on_the_wire(self, client):
+        # Series identity str()s keys, but the wire must too — OTLP KeyValue.key is a
+        # string field and strict decoders reject numeric keys.
+        client.metrics.count("m", 1, attributes={"a": 1, 2: "x"})
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        (dp,) = metric["sum"]["dataPoints"]
+        keys = {attr["key"] for attr in dp["attributes"]}
+        assert keys == {"a", "2"}
+        assert all(isinstance(attr["key"], str) for attr in dp["attributes"])
+
+    def test_list_attribute_records_as_array_value(self, client):
+        client.metrics.count("arr", 1, attributes={"tags": ["a", "b"]})
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        (dp,) = metric["sum"]["dataPoints"]
+        (attr,) = dp["attributes"]
+        assert attr["value"] == {
+            "arrayValue": {"values": [{"stringValue": "a"}, {"stringValue": "b"}]}
+        }
+
+    @pytest.mark.parametrize(
+        "hook",
+        [
+            lambda s: True,  # truthy non-dict return
+            lambda s: {**s, "type": "guage"},  # typo'd type must not fold as histogram
+        ],
+    )
+    def test_misbehaving_before_send_drops_sample(self, hook):
+        c = Client(
+            FAKE_API_KEY,
+            host="https://us.example.com",
+            sync_mode=True,
+            metrics={"before_send": hook},
+        )
+        c.metrics.count("m", 1)
+
+        payload, _, _ = flush_and_capture(c)
+        c.metrics.reset()
+
+        assert payload is None
+
+    def test_bool_and_int_attribute_values_are_distinct_series(self, client):
+        # Python True == 1 collapsed these into one series with a tuple-based key.
+        client.metrics.count("m", 1, attributes={"k": True})
+        client.metrics.count("m", 1, attributes={"k": 1})
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        assert len(metric["sum"]["dataPoints"]) == 2
+
+    def test_none_attribute_values_dropped_before_keying(self, client):
+        # A None-valued attribute split the series key but was stripped from the wire,
+        # emitting two indistinguishable data points.
+        client.metrics.count("m", 1, attributes={"k": None})
+        client.metrics.count("m", 1)
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        (dp,) = metric["sum"]["dataPoints"]
+        assert dp["asDouble"] == 2.0
+
+
+class TestMetricsEncodingParity:
+    # The two SDKs must emit byte-identical AnyValues for the same logical input.
+
+    @pytest.mark.parametrize(
+        "value,expected",
+        [
+            (
+                math.inf,
+                {"stringValue": "Infinity"},
+            ),  # proto3 JSON literal, not Python's "inf"
+            (-math.inf, {"stringValue": "-Infinity"}),
+            (math.nan, {"stringValue": "NaN"}),
+            (
+                2.0,
+                {"intValue": 2},
+            ),  # integral floats encode as intValue, matching JS Number.isInteger
+            (2.5, {"doubleValue": 2.5}),
+        ],
+    )
+    def test_any_value_encoding_matches_js(self, client, value, expected):
+        client.metrics.count("enc", 1, attributes={"v": value})
+
+        payload, _, _ = flush_and_capture(client)
+
+        (metric,) = metrics_from(payload)
+        (dp,) = metric["sum"]["dataPoints"]
+        (attr,) = dp["attributes"]
+        assert attr["value"] == expected
+
+
+class TestMetricsFailureHandling:
+    def test_fork_drops_inherited_window_and_rearms(self, client):
+        # A forked child inherits a dead timer thread and the parent's window; without the
+        # PID guard it never flushes again and duplicates the parent's samples.
+        client.metrics.count("m", 1)
+        assert client.metrics._flush_timer is not None
+        client.metrics._pid -= 1  # simulate being in a fork child
+
+        client.metrics.count("m", 5)
+
+        payload, _, _ = flush_and_capture(client)
+        (metric,) = metrics_from(payload)
+        (dp,) = metric["sum"]["dataPoints"]
+        assert (
+            dp["asDouble"] == 5.0
+        )  # inherited window dropped, only the child's sample remains
+
+    def test_merge_back_respects_series_cap(self):
+        c = Client(
+            FAKE_API_KEY,
+            host="https://us.example.com",
+            sync_mode=True,
+            metrics={"max_series_per_flush": 2},
+        )
+        c.metrics.count("m", 1, attributes={"k": "a"})
+        c.metrics.count("m", 1, attributes={"k": "b"})
+        with mock.patch(
+            "posthog.metrics_capture._get_session", return_value=mock_session(503)
+        ):
+            c.metrics.flush()
+        # Attribute churn: two NEW series in the fresh window, then the failed window merges back.
+        c.metrics.count("m", 1, attributes={"k": "c"})
+        c.metrics.count("m", 1, attributes={"k": "d"})
+
+        assert (
+            len(c.metrics._series) <= 2
+        )  # cap holds through merge-back; no unbounded backlog
+        c.metrics.reset()
+
+    def test_window_dropped_after_consecutive_failures(self, client):
+        # A permanently-down endpoint must not buffer forever: after the retry budget the
+        # window is dropped loudly instead of growing until a 413 destroys everything.
+        client.metrics.count("m", 3)
+        with mock.patch(
+            "posthog.metrics_capture._get_session", return_value=mock_session(503)
+        ):
+            for _ in range(4):
+                client.metrics.flush()
+
+        payload, _, _ = flush_and_capture(client)
+
+        assert (
+            payload is None
+        )  # budget exhausted → window dropped, nothing left to send
diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt
index 60072517..05256028 100644
--- a/references/public_api_snapshot.txt
+++ b/references/public_api_snapshot.txt
@@ -244,6 +244,7 @@ alias posthog.client.InconclusiveMatchError -> posthog.feature_flags.Inconclusiv
 alias posthog.client.OptionalCaptureArgs -> posthog.args.OptionalCaptureArgs
 alias posthog.client.OptionalSetArgs -> posthog.args.OptionalSetArgs
 alias posthog.client.Poller -> posthog.poller.Poller
+alias posthog.client.PostHogMetrics -> posthog.metrics_capture.PostHogMetrics
 alias posthog.client.QuotaLimitError -> posthog.request.QuotaLimitError
 alias posthog.client.RedisFlagCache -> posthog.utils.RedisFlagCache
 alias posthog.client.RequestsConnectionError -> posthog.request.RequestsConnectionError
@@ -332,6 +333,8 @@ alias posthog.mcp.__version__ -> posthog.mcp.version.__version__
 alias posthog.mcp.derive_session_id_from_mcp_session -> posthog.mcp.session.derive_session_id_from_mcp_session
 alias posthog.mcp.get_more_tools_result -> posthog.mcp.tools.get_more_tools_result
 alias posthog.mcp.set_logger -> posthog.mcp.logger.set_logger
+alias posthog.metrics_capture.VERSION -> posthog.version.VERSION
+alias posthog.metrics_capture.remove_trailing_slash -> posthog.utils.remove_trailing_slash
 alias posthog.request.VERSION -> posthog.version.VERSION
 alias posthog.request.remove_trailing_slash -> posthog.utils.remove_trailing_slash
 alias posthog.set_socket_options -> posthog.request.set_socket_options
@@ -526,6 +529,7 @@ attribute posthog.client.Client.is_server = is_server
 attribute posthog.client.Client.log = logging.getLogger('posthog')
 attribute posthog.client.Client.log_captured_exceptions = log_captured_exceptions
 attribute posthog.client.Client.max_retries = max_retries
+attribute posthog.client.Client.metrics: PostHogMetrics
 attribute posthog.client.Client.on_error = on_error
 attribute posthog.client.Client.personal_api_key = self.secret_key
 attribute posthog.client.Client.poll_interval = poll_interval
@@ -701,6 +705,9 @@ attribute posthog.mcp.types.UserIdentity.distinct_id: str
 attribute posthog.mcp.types.UserIdentity.groups: Optional[Dict[str, str]] = None
 attribute posthog.mcp.types.UserIdentity.properties: Optional[JsonRecord] = None
 attribute posthog.mcp.version.__version__ = '0.1.0'
+attribute posthog.metrics_capture.DEFAULT_HISTOGRAM_BOUNDS = [0, 5, 10, 25, 50, 75, 100, 250, 500, 750, 1000, 2500, 5000, 7500, 10000]
+attribute posthog.metrics_capture.MetricAttributeValue = Union[str, int, float, bool]
+attribute posthog.metrics_capture.log = logging.getLogger('posthog')
 attribute posthog.on_error = None
 attribute posthog.personal_api_key = None
 attribute posthog.poll_interval = 30
@@ -855,7 +862,7 @@ class posthog.bucketed_rate_limiter.BucketedRateLimiter(bucket_size: Number, ref
 class posthog.capture_compression.CaptureCompression 
 class posthog.capture_mode.CaptureMode 
 class posthog.capture_v1.CaptureV1Error(status: int | str, message: str, *, retry_after: Optional[float] = None, request_id: Optional[str] = None, attempts: Optional[int] = None, retry_exhausted: Optional[list[str]] = None, drops: Optional[list[tuple[str, Optional[str]]]] = None)
-class posthog.client.Client(project_api_key: str, host=None, debug=False, max_queue_size=10000, send=True, on_error=None, flush_at=100, flush_interval=5.0, gzip=False, max_retries=3, sync_mode=False, timeout=15, thread=1, poll_interval=30, personal_api_key=None, disabled=False, disable_geoip=True, is_server=True, historical_migration=False, feature_flags_request_timeout_seconds=3, feature_flags_request_max_retries=1, super_properties=None, enable_exception_autocapture=False, log_captured_exceptions=False, project_root=None, privacy_mode=False, before_send=None, flag_fallback_cache_url=None, enable_local_evaluation=True, flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None, capture_exception_code_variables=False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: list[str] | None = None, enable_exception_autocapture_rate_limiting=False, exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, capture_mode: Optional[Union[CaptureMode, str]] = None, capture_compression: Optional[Union[CaptureCompression, str]] = None, secret_key=None, _dedicated_ai_endpoint=False)
+class posthog.client.Client(project_api_key: str, host=None, debug=False, max_queue_size=10000, send=True, on_error=None, flush_at=100, flush_interval=5.0, gzip=False, max_retries=3, sync_mode=False, timeout=15, thread=1, poll_interval=30, personal_api_key=None, disabled=False, disable_geoip=True, is_server=True, historical_migration=False, feature_flags_request_timeout_seconds=3, feature_flags_request_max_retries=1, super_properties=None, enable_exception_autocapture=False, log_captured_exceptions=False, project_root=None, privacy_mode=False, before_send=None, flag_fallback_cache_url=None, enable_local_evaluation=True, flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None, capture_exception_code_variables=False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: list[str] | None = None, enable_exception_autocapture_rate_limiting=False, exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, capture_mode: Optional[Union[CaptureMode, str]] = None, capture_compression: Optional[Union[CaptureCompression, str]] = None, secret_key=None, _dedicated_ai_endpoint=False, metrics: Optional[dict] = None)
 class posthog.consumer.Consumer(queue, api_key, flush_at=100, host=None, on_error=None, flush_interval=5.0, gzip=False, retries=10, timeout=15, historical_migration=False, dedicated_ai_endpoint=False, capture_mode=CaptureMode.V0, capture_compression=CaptureCompression.NONE)
 class posthog.contexts.ContextScope(parent=None, fresh: bool = False, capture_exceptions: bool = True, client: Optional[Client] = None)
 class posthog.exception_capture.ExceptionCapture(client: Client, rate_limiting_enabled=False, bucket_size=DEFAULT_BUCKET_SIZE, refill_rate=DEFAULT_REFILL_RATE, refill_interval_seconds=DEFAULT_REFILL_INTERVAL_SECONDS)
@@ -878,6 +885,7 @@ class posthog.mcp.types.MCPAnalyticsContextOptions(description: Optional[str] =
 class posthog.mcp.types.MCPAnalyticsOptions(logger: Optional[LoggerFn] = None, report_missing: bool = False, missing_capability_tool_name: Optional[str] = None, enable_conversation_id: bool = False, enable_exception_autocapture: bool = True, context: Union[bool, MCPAnalyticsContextOptions] = True, identify: Optional[Union[IdentifyFn, UserIdentity]] = None, intent_fallback: Optional[IntentFallbackFn] = None, before_send: Optional[BeforeSendFn] = None, event_properties: Optional[EventPropertiesFn] = None)
 class posthog.mcp.types.PreparedToolCall(args: Optional[JsonRecord] = None, intent: Optional[str] = None, intent_source: Optional[str] = None, is_missing_capability: bool = False)
 class posthog.mcp.types.UserIdentity(distinct_id: str, properties: Optional[JsonRecord] = None, groups: Optional[Dict[str, str]] = None)
+class posthog.metrics_capture.PostHogMetrics(client, config: Optional[dict] = None)
 class posthog.poller.Poller(interval, execute, *args, **kwargs)
 class posthog.request.APIError(status: Union[int, str], message: str, retry_after: Optional[float] = None)
 class posthog.request.DatetimeSerializer 
@@ -1264,6 +1272,11 @@ method posthog.mcp.posthog_mcp.PostHogMCP.flush(timeout_seconds: Optional[float]
 method posthog.mcp.posthog_mcp.PostHogMCP.prepare_tool_call(name: str, args: Optional[JsonRecord] = None) -> PreparedToolCall
 method posthog.mcp.posthog_mcp.PostHogMCP.prepare_tool_list(tools: List[Any], context: Union[bool, MCPAnalyticsContextOptions] = True, report_missing: bool = False) -> List[Any]
 method posthog.mcp.posthog_mcp.PostHogMCP.shutdown() -> None
+method posthog.metrics_capture.PostHogMetrics.count(name: str, value: float = 1, unit: Optional[str] = None, attributes: Optional[dict] = None) -> None
+method posthog.metrics_capture.PostHogMetrics.flush() -> None
+method posthog.metrics_capture.PostHogMetrics.gauge(name: str, value: float, unit: Optional[str] = None, attributes: Optional[dict] = None) -> None
+method posthog.metrics_capture.PostHogMetrics.histogram(name: str, value: float, unit: Optional[str] = None, attributes: Optional[dict] = None) -> None
+method posthog.metrics_capture.PostHogMetrics.reset() -> None
 method posthog.poller.Poller.run()
 method posthog.poller.Poller.stop()
 method posthog.request.DatetimeSerializer.default(obj: Any)
@@ -1347,6 +1360,7 @@ module posthog.mcp.session
 module posthog.mcp.tools
 module posthog.mcp.types
 module posthog.mcp.version
+module posthog.metrics_capture
 module posthog.poller
 module posthog.request
 module posthog.types
diff --git a/typings/requests/exceptions.pyi b/typings/requests/exceptions.pyi
index 73ffa5cc..f7311f21 100644
--- a/typings/requests/exceptions.pyi
+++ b/typings/requests/exceptions.pyi
@@ -1,2 +1,3 @@
-class Timeout(Exception): ...
-class ConnectionError(Exception): ...
+class RequestException(Exception): ...
+class Timeout(RequestException): ...
+class ConnectionError(RequestException): ...