diff --git a/.sampo/changesets/on-drop-callback.md b/.sampo/changesets/on-drop-callback.md
new file mode 100644
index 00000000..c3d38d77
--- /dev/null
+++ b/.sampo/changesets/on-drop-callback.md
@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Add an on_drop callback invoked with the event when the internal queue is full, so callers can observe or count dropped events. The SDK always drops rather than blocking the caller.
diff --git a/posthog/client.py b/posthog/client.py
index 99ef4801..9fcb1520 100644
--- a/posthog/client.py
+++ b/posthog/client.py
@@ -260,6 +260,7 @@ def __init__(
         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,
+        on_drop=None,
         _dedicated_ai_endpoint=False,
     ):
         """
@@ -343,6 +344,10 @@ def __init__(
                 interval for each exception type's bucket.
             exception_autocapture_refill_interval_seconds: Seconds between
                 token refills for autocaptured exception rate limiting.
+            on_drop: Optional callback invoked with the dropped event when the
+                internal queue is full. The SDK always drops rather than blocking
+                the caller, so this is how you observe or count dropped events.
+                Runs on the calling thread, so keep it fast and non-blocking.
 
         Examples:
             ```python
@@ -354,6 +359,10 @@ def __init__(
         Category:
             Initialization
         """
+        if on_drop is not None and not callable(on_drop):
+            raise ValueError("on_drop must be callable")
+        self.on_drop = on_drop
+
         self._max_queue_size = max_queue_size
         self.queue: Queue = Queue(max_queue_size)
 
@@ -1647,6 +1656,11 @@ def _enqueue(self, msg, disable_geoip):
             return sent_uuid
         except Full:
             self.log.warning("analytics-python queue is full")
+            if self.on_drop is not None:
+                try:
+                    self.on_drop(msg)
+                except Exception:
+                    self.log.exception("on_drop callback raised")
             return None
 
     def flush(self, timeout_seconds: Optional[float] = 10) -> None:
diff --git a/posthog/test/test_client.py b/posthog/test/test_client.py
index 98bb4914..a3ef2152 100644
--- a/posthog/test/test_client.py
+++ b/posthog/test/test_client.py
@@ -2092,6 +2092,40 @@ def test_overflow(self):
         # Make sure we are informed that the queue is at capacity
         self.assertIsNone(msg_uuid)
 
+    def test_on_drop_called_when_queue_full(self):
+        dropped = []
+        client = Client(FAKE_TEST_API_KEY, max_queue_size=1, on_drop=dropped.append)
+        client.join()
+        first_uuid = client.capture("first event", distinct_id="distinct_id")
+        self.assertIsNotNone(first_uuid)
+
+        second_uuid = client.capture("second event", distinct_id="distinct_id")
+        self.assertIsNone(second_uuid)
+        self.assertEqual(len(dropped), 1)
+        self.assertEqual(dropped[0]["event"], "second event")
+
+    def test_on_drop_not_called_when_space_available(self):
+        dropped = []
+        client = Client(FAKE_TEST_API_KEY, max_queue_size=10, on_drop=dropped.append)
+        client.join()
+        first_uuid = client.capture("first event", distinct_id="distinct_id")
+        self.assertIsNotNone(first_uuid)
+        self.assertEqual(dropped, [])
+
+    def test_on_drop_callback_exception_is_swallowed(self):
+        def boom(_msg):
+            raise RuntimeError("boom")
+
+        client = Client(FAKE_TEST_API_KEY, max_queue_size=1, on_drop=boom)
+        client.join()
+        client.capture("first event", distinct_id="distinct_id")
+        second_uuid = client.capture("second event", distinct_id="distinct_id")
+        self.assertIsNone(second_uuid)
+
+    def test_on_drop_invalid_config(self):
+        with self.assertRaises(ValueError):
+            Client(FAKE_TEST_API_KEY, on_drop="not callable")
+
     def test_unicode(self):
         Client("unicode_key")