diff --git a/python_anticaptcha/base.py b/python_anticaptcha/base.py
index 7db0c38..dee0b96 100644
--- a/python_anticaptcha/base.py
+++ b/python_anticaptcha/base.py
@@ -67,12 +67,23 @@ def __repr__(self) -> str:
             return f"<Job task_id={self.task_id} status={status!r}>"
         return f"<Job task_id={self.task_id}>"
 
-    def join(self, maximum_time: int | None = None) -> None:
+    def join(self, maximum_time: int | None = None, on_check=None) -> None:
+        """Poll for task completion, blocking until ready or timeout.
+
+        :param maximum_time: Maximum seconds to wait (default: ``MAXIMUM_JOIN_TIME``).
+        :param on_check: Optional callback invoked after each poll with
+            ``(elapsed_time, status)`` where *elapsed_time* is the total seconds
+            waited so far and *status* is the last task status string
+            (e.g. ``"processing"``).
+        :raises AnticaptchaException: If *maximum_time* is exceeded.
+        """
         elapsed_time = 0
         maximum_time = maximum_time or MAXIMUM_JOIN_TIME
         while not self.check_is_ready():
             time.sleep(SLEEP_EVERY_CHECK_FINISHED)
             elapsed_time += SLEEP_EVERY_CHECK_FINISHED
+            if on_check is not None:
+                on_check(elapsed_time, self._last_result.get("status"))
             if elapsed_time > maximum_time:
                 raise AnticaptchaException(
                     None,
diff --git a/tests/test_base.py b/tests/test_base.py
index 71afd6b..5c97223 100644
--- a/tests/test_base.py
+++ b/tests/test_base.py
@@ -166,6 +166,42 @@ def test_timeout_raises(self, mock_sleep):
         assert "exceeded" in str(exc_info.value).lower()
 
 
+class TestJobJoinOnCheck:
+    @patch("python_anticaptcha.base.time.sleep")
+    def test_on_check_called_each_iteration(self, mock_sleep):
+        client = MagicMock()
+        # Return processing twice, then ready
+        client.getTaskResult.side_effect = [
+            {"status": "processing"},
+            {"status": "processing"},
+            {"status": "ready", "solution": {}},
+        ]
+        job = Job(client, task_id=1)
+        callback = MagicMock()
+        job.join(on_check=callback)
+        assert callback.call_count == 2
+        # Verify elapsed time and status passed correctly
+        callback.assert_any_call(SLEEP_EVERY_CHECK_FINISHED, "processing")
+        callback.assert_any_call(SLEEP_EVERY_CHECK_FINISHED * 2, "processing")
+
+    @patch("python_anticaptcha.base.time.sleep")
+    def test_on_check_none_by_default(self, mock_sleep):
+        client = MagicMock()
+        client.getTaskResult.return_value = {"status": "ready", "solution": {}}
+        job = Job(client, task_id=1)
+        # Should not raise when on_check is not provided
+        job.join()
+
+    @patch("python_anticaptcha.base.time.sleep")
+    def test_on_check_not_called_when_immediately_ready(self, mock_sleep):
+        client = MagicMock()
+        client.getTaskResult.return_value = {"status": "ready", "solution": {}}
+        job = Job(client, task_id=1)
+        callback = MagicMock()
+        job.join(on_check=callback)
+        callback.assert_not_called()
+
+
 class TestContextManager:
     def test_enter_returns_self(self):
         client = AnticaptchaClient("key123")