Python CAPTCHA Solving with Retry and Error Handling Patterns

Production CAPTCHA solving fails. Networks drop, APIs time out, tokens expire. This guide covers every error pattern you'll hit with CaptchaAI and the retry strategies that keep your pipeline running.

CaptchaAI error classification

Retriable errors (retry immediately or with backoff)

Non-retriable errors (fix the input)

Basic retry with exponential backoff

import time
import requests

API_KEY = "YOUR_API_KEY"

# Define error categories
RETRIABLE_ERRORS = {
    "ERROR_NO_SLOT_AVAILABLE",
    "CAPCHA_NOT_READY",
}

FATAL_ERRORS = {
    "ERROR_WRONG_USER_KEY",
    "ERROR_KEY_DOES_NOT_EXIST",
    "ERROR_ZERO_BALANCE",
    "ERROR_CAPTCHA_UNSOLVABLE",
    "ERROR_BAD_DUPLICATES",
    "ERROR_BAD_PARAMETERS",
    "ERROR_WRONG_CAPTCHA_ID",
}


class CaptchaError(Exception):
    """Base CAPTCHA error."""
    def __init__(self, code, message=""):
        self.code = code
        super().__init__(f"{code}: {message}")


class RetriableError(CaptchaError):
    """Error that can be retried."""
    pass


class FatalError(CaptchaError):
    """Error that should not be retried."""
    pass


def classify_error(error_code):
    """Classify an error as retriable or fatal."""
    if error_code in RETRIABLE_ERRORS:
        raise RetriableError(error_code)
    elif error_code in FATAL_ERRORS:
        raise FatalError(error_code)
    else:
        # Unknown errors — treat as retriable
        raise RetriableError(error_code, "Unknown error")

Retry decorator

import functools
import random


def retry_captcha(max_retries=3, base_delay=2, max_delay=30, jitter=True):
    """Decorator for retrying CAPTCHA operations with exponential backoff."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)

                except FatalError:
                    raise  # Don't retry fatal errors

                except RetriableError as e:
                    last_exception = e
                    if attempt < max_retries:
                        delay = min(base_delay * (2 ** attempt), max_delay)
                        if jitter:
                            delay *= (0.5 + random.random())
                        print(f"Retry {attempt + 1}/{max_retries} after {delay:.1f}s: {e}")
                        time.sleep(delay)

                except requests.exceptions.RequestException as e:
                    last_exception = e
                    if attempt < max_retries:
                        delay = min(base_delay * (2 ** attempt), max_delay)
                        print(f"Network retry {attempt + 1}/{max_retries}: {e}")
                        time.sleep(delay)

            raise last_exception

        return wrapper
    return decorator

Production solver with full error handling

class RobustCaptchaSolver:
    """Production CAPTCHA solver with retry, backoff, and error classification."""

    def __init__(self, api_key, max_retries=3, poll_interval=5, max_poll_time=150):
        self.api_key = api_key
        self.max_retries = max_retries
        self.poll_interval = poll_interval
        self.max_poll_time = max_poll_time

    @retry_captcha(max_retries=3, base_delay=2)
    def solve(self, method, **params):
        """Solve a CAPTCHA with full error handling and retry."""
        task_id = self._submit(method, **params)
        return self._poll(task_id)

    def _submit(self, method, **params):
        """Submit task with retry on network and slot errors."""
        for attempt in range(self.max_retries + 1):
            try:
                resp = requests.post("https://ocr.captchaai.com/in.php", data={
                    "key": self.api_key, "method": method, "json": 1, **params,
                }, timeout=30)
                resp.raise_for_status()
                data = resp.json()

                if data.get("status") == 1:
                    return data["request"]

                error_code = data.get("request", "UNKNOWN")

                if error_code == "ERROR_NO_SLOT_AVAILABLE":
                    if attempt < self.max_retries:
                        delay = 3 * (attempt + 1)
                        print(f"No slot available, waiting {delay}s...")
                        time.sleep(delay)
                        continue
                    raise RetriableError(error_code)

                if error_code in FATAL_ERRORS:
                    raise FatalError(error_code)

                raise RetriableError(error_code)

            except requests.exceptions.RequestException as e:
                if attempt < self.max_retries:
                    time.sleep(2 ** attempt)
                    continue
                raise

        raise RetriableError("MAX_SUBMIT_RETRIES")

    def _poll(self, task_id):
        """Poll for result with timeout."""
        start = time.time()
        attempts = 0

        while time.time() - start < self.max_poll_time:
            time.sleep(self.poll_interval)
            attempts += 1

            try:
                resp = requests.get("https://ocr.captchaai.com/res.php", params={
                    "key": self.api_key, "action": "get", "id": task_id, "json": 1,
                }, timeout=30)
                resp.raise_for_status()
                data = resp.json()

                if data.get("status") == 1:
                    print(f"Solved after {attempts} polls ({time.time() - start:.1f}s)")
                    return data["request"]

                error_code = data.get("request", "")

                if error_code == "CAPCHA_NOT_READY":
                    continue

                if error_code in FATAL_ERRORS:
                    raise FatalError(error_code)

                # Unknown poll error — keep polling
                continue

            except requests.exceptions.RequestException:
                # Network error during poll — keep trying
                continue

        raise TimeoutError(f"Solve timed out after {self.max_poll_time}s ({attempts} polls)")


# Usage
solver = RobustCaptchaSolver(API_KEY)

try:
    token = solver.solve("userrecaptcha", googlekey="SITEKEY", pageurl="https://example.com")
    print(f"Token: {token[:50]}...")
except FatalError as e:
    print(f"Fatal: {e}")
except RetriableError as e:
    print(f"All retries exhausted: {e}")
except TimeoutError as e:
    print(f"Timeout: {e}")

Circuit breaker pattern

Prevent cascading failures when the API is down:

import time


class CircuitBreaker:
    """Stops calling the API when too many errors occur."""

    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.last_failure_time = 0
        self.state = "closed"  # closed=normal, open=blocked, half-open=testing

    def can_execute(self):
        if self.state == "closed":
            return True

        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
                return True
            return False

        # half-open — allow one test request
        return True

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()

        if self.failures >= self.failure_threshold:
            self.state = "open"
            print(f"Circuit OPEN — pausing for {self.recovery_timeout}s")


class CircuitBreakerSolver:
    """Solver with circuit breaker protection."""

    def __init__(self, api_key):
        self.api_key = api_key
        self.breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

    def solve(self, method, **params):
        if not self.breaker.can_execute():
            raise Exception("Circuit breaker open — API appears to be down")

        try:
            result = self._do_solve(method, **params)
            self.breaker.record_success()
            return result
        except FatalError:
            raise
        except Exception as e:
            self.breaker.record_failure()
            raise

    def _do_solve(self, method, **params):
        submit = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }, timeout=30).json()

        if submit.get("status") != 1:
            error_code = submit.get("request", "UNKNOWN")
            if error_code in FATAL_ERRORS:
                raise FatalError(error_code)
            raise RetriableError(error_code)

        task_id = submit["request"]
        for _ in range(30):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }, timeout=30).json()
            if result.get("status") == 1:
                return result["request"]
            if result.get("request") in FATAL_ERRORS:
                raise FatalError(result["request"])

        raise TimeoutError("Timed out")

Token expiration handling

CAPTCHA tokens expire (reCAPTCHA: ~2 minutes, Turnstile: ~5 minutes). Handle this:

import time


class TokenManager:
    """Manage CAPTCHA token lifecycle — solve, cache, detect expiry."""

    def __init__(self, solver, default_ttl=110):
        self.solver = solver
        self.default_ttl = default_ttl
        self.cache = {}  # key -> (token, timestamp)

    def get_token(self, cache_key, method, **params):
        """Get a valid token, solving only if needed."""
        if cache_key in self.cache:
            token, timestamp = self.cache[cache_key]
            age = time.time() - timestamp
            if age < self.default_ttl:
                return token
            print(f"Token expired ({age:.0f}s old), re-solving...")

        token = self.solver.solve(method, **params)
        self.cache[cache_key] = (token, time.time())
        return token

    def invalidate(self, cache_key):
        """Mark token as invalid (e.g., server rejected it)."""
        self.cache.pop(cache_key, None)

    def solve_with_expiry_retry(self, method, submit_fn, max_attempts=2, **params):
        """Solve and submit, re-solving if server rejects the token."""
        for attempt in range(max_attempts):
            token = self.solver.solve(method, **params)
            success = submit_fn(token)
            if success:
                return token
            print(f"Token rejected (attempt {attempt + 1}), re-solving...")

        raise Exception("Token rejected after max attempts")

Logging and monitoring

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("captcha_solver")


class LoggingSolver:
    """Solver wrapper with structured logging."""

    def __init__(self, solver):
        self.solver = solver
        self.stats = {"solved": 0, "failed": 0, "retries": 0}

    def solve(self, method, **params):
        start = time.time()
        url = params.get("pageurl", "unknown")

        try:
            token = self.solver.solve(method, **params)
            elapsed = time.time() - start
            self.stats["solved"] += 1
            logger.info(f"Solved {method} for {url} in {elapsed:.1f}s")
            return token

        except FatalError as e:
            self.stats["failed"] += 1
            logger.error(f"Fatal error for {url}: {e}")
            raise

        except Exception as e:
            self.stats["failed"] += 1
            logger.warning(f"Failed for {url}: {e}")
            raise

    def report(self):
        total = self.stats["solved"] + self.stats["failed"]
        rate = (self.stats["solved"] / total * 100) if total else 0
        logger.info(f"Stats: {self.stats['solved']}/{total} solved ({rate:.1f}%)")

Complete production pattern

# Combine all patterns
solver = LoggingSolver(
    CircuitBreakerSolver(API_KEY)
)
token_mgr = TokenManager(solver, default_ttl=110)

# Usage
try:
    token = token_mgr.get_token(
        "login_page",
        "userrecaptcha",
        googlekey="SITEKEY",
        pageurl="https://example.com/login",
    )
    print(f"Token: {token[:50]}...")
except FatalError as e:
    print(f"Cannot solve: {e}")
except Exception as e:
    print(f"Temporary failure: {e}")
finally:
    solver.report()

Troubleshooting

Frequently asked questions

Should I retry ERROR_CAPTCHA_UNSOLVABLE?

No. This means CaptchaAI's workers couldn't solve the CAPTCHA. Retrying the same CAPTCHA will likely fail again. Try with different parameters or skip.

What's a good max retry count?

3 retries for submission, 30 polls for result checking. More than 3 submission retries usually means a deeper issue (wrong key, zero balance).

How do I handle balance running out mid-batch?

Check your balance before starting: GET /res.php?key=KEY&action=getbalance. If balance is low, stop submitting new tasks.

Summary

Robust CAPTCHA solving with CaptchaAI requires proper error classification, exponential backoff, circuit breakers, and token expiration handling. Use the RobustCaptchaSolver + CircuitBreaker + TokenManager stack for production reliability.

Related Articles

• Captchaai Callback Error Handling Patterns
• Nodejs Captcha Solving Retry Error Handling