Building a Python Wrapper Library for CaptchaAI API

A wrapper library centralizes API interaction, error handling, and retry logic so every script in your project uses consistent, tested code. This guide builds a production-ready Python wrapper for CaptchaAI step by step.

Project structure

captchaai_client/
├── __init__.py
├── client.py
├── exceptions.py
└── models.py

Exceptions

# captchaai_client/exceptions.py

class CaptchaAIError(Exception):
    """Base exception for CaptchaAI errors."""
    def __init__(self, code: str, message: str = ""):
        self.code = code
        super().__init__(f"{code}: {message}" if message else code)


class WrongAPIKeyError(CaptchaAIError):
    pass


class ZeroBalanceError(CaptchaAIError):
    pass


class TaskTimeoutError(CaptchaAIError):
    pass


ERROR_MAP = {
    "ERROR_WRONG_USER_KEY": WrongAPIKeyError,
    "ERROR_KEY_DOES_NOT_EXIST": WrongAPIKeyError,
    "ERROR_ZERO_BALANCE": ZeroBalanceError,
}


def raise_for_error(code: str):
    exc_class = ERROR_MAP.get(code, CaptchaAIError)
    raise exc_class(code)

Models

# captchaai_client/models.py

from dataclasses import dataclass
from typing import Optional


@dataclass
class SolveResult:
    task_id: str
    token: str
    solve_time: float  # seconds


@dataclass
class RecaptchaV2Params:
    sitekey: str
    page_url: str
    invisible: bool = False
    data_s: Optional[str] = None
    proxy: Optional[str] = None
    proxy_type: Optional[str] = None


@dataclass
class RecaptchaV3Params:
    sitekey: str
    page_url: str
    action: str = "verify"
    min_score: float = 0.3


@dataclass
class TurnstileParams:
    sitekey: str
    page_url: str
    action: Optional[str] = None
    cdata: Optional[str] = None


@dataclass
class ImageParams:
    body: str  # base64-encoded image
    phrase: bool = False
    case_sensitive: bool = False
    numeric: int = 0  # 0=any, 1=numeric only
    min_len: int = 0
    max_len: int = 0

Client

# captchaai_client/client.py

import time
import requests
from typing import Union
from .exceptions import CaptchaAIError, TaskTimeoutError, raise_for_error
from .models import (
    SolveResult,
    RecaptchaV2Params,
    RecaptchaV3Params,
    TurnstileParams,
    ImageParams,
)

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"


class CaptchaAI:
    def __init__(
        self,
        api_key: str,
        timeout: int = 120,
        poll_interval: int = 5,
        max_retries: int = 2,
    ):
        self.api_key = api_key
        self.timeout = timeout
        self.poll_interval = poll_interval
        self.max_retries = max_retries
        self._session = requests.Session()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self._session.close()

    def get_balance(self) -> float:
        resp = self._session.get(RESULT_URL, params={
            "key": self.api_key,
            "action": "getbalance",
            "json": "1",
        })
        data = resp.json()
        if data["status"] != 1:
            raise_for_error(data["request"])
        return float(data["request"])

    def solve_recaptcha_v2(self, params: RecaptchaV2Params) -> SolveResult:
        form = {
            "method": "userrecaptcha",
            "googlekey": params.sitekey,
            "pageurl": params.page_url,
        }
        if params.invisible:
            form["invisible"] = "1"
        if params.data_s:
            form["data-s"] = params.data_s
        if params.proxy:
            form["proxy"] = params.proxy
            form["proxytype"] = params.proxy_type or "HTTP"
        return self._solve(form)

    def solve_recaptcha_v3(self, params: RecaptchaV3Params) -> SolveResult:
        return self._solve({
            "method": "userrecaptcha",
            "version": "v3",
            "googlekey": params.sitekey,
            "pageurl": params.page_url,
            "action": params.action,
            "min_score": str(params.min_score),
        })

    def solve_turnstile(self, params: TurnstileParams) -> SolveResult:
        form = {
            "method": "turnstile",
            "sitekey": params.sitekey,
            "pageurl": params.page_url,
        }
        if params.action:
            form["action"] = params.action
        if params.cdata:
            form["data"] = params.cdata
        return self._solve(form)

    def solve_image(self, params: ImageParams) -> SolveResult:
        form = {
            "method": "base64",
            "body": params.body,
        }
        if params.phrase:
            form["phrase"] = "1"
        if params.case_sensitive:
            form["regsense"] = "1"
        if params.numeric:
            form["numeric"] = str(params.numeric)
        if params.min_len:
            form["min_len"] = str(params.min_len)
        if params.max_len:
            form["max_len"] = str(params.max_len)
        return self._solve(form)

    def report_bad(self, task_id: str) -> None:
        self._session.get(RESULT_URL, params={
            "key": self.api_key,
            "action": "reportbad",
            "id": task_id,
            "json": "1",
        })

    def report_good(self, task_id: str) -> None:
        self._session.get(RESULT_URL, params={
            "key": self.api_key,
            "action": "reportgood",
            "id": task_id,
            "json": "1",
        })

    def _solve(self, form: dict) -> SolveResult:
        form["key"] = self.api_key
        form["json"] = "1"

        for attempt in range(self.max_retries + 1):
            try:
                return self._submit_and_poll(form)
            except CaptchaAIError as e:
                if attempt == self.max_retries:
                    raise
                if e.code in ("ERROR_NO_SLOT_AVAILABLE",):
                    time.sleep(2 ** attempt)
                else:
                    raise

    def _submit_and_poll(self, form: dict) -> SolveResult:
        start = time.time()

        resp = self._session.post(SUBMIT_URL, data=form)
        data = resp.json()
        if data["status"] != 1:
            raise_for_error(data["request"])

        task_id = data["request"]
        deadline = start + self.timeout

        while time.time() < deadline:
            time.sleep(self.poll_interval)
            poll_resp = self._session.get(RESULT_URL, params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": "1",
            })
            poll_data = poll_resp.json()
            if poll_data["status"] == 1:
                return SolveResult(
                    task_id=task_id,
                    token=poll_data["request"],
                    solve_time=time.time() - start,
                )
            if poll_data["request"] != "CAPCHA_NOT_READY":
                raise_for_error(poll_data["request"])

        raise TaskTimeoutError("TIMEOUT", f"Task {task_id} timed out")

Package init

# captchaai_client/__init__.py

from .client import CaptchaAI
from .models import (
    RecaptchaV2Params,
    RecaptchaV3Params,
    TurnstileParams,
    ImageParams,
    SolveResult,
)
from .exceptions import CaptchaAIError, WrongAPIKeyError, ZeroBalanceError, TaskTimeoutError

__all__ = [
    "CaptchaAI",
    "RecaptchaV2Params",
    "RecaptchaV3Params",
    "TurnstileParams",
    "ImageParams",
    "SolveResult",
    "CaptchaAIError",
    "WrongAPIKeyError",
    "ZeroBalanceError",
    "TaskTimeoutError",
]

Usage examples

Basic usage

from captchaai_client import CaptchaAI, RecaptchaV2Params

with CaptchaAI(api_key="YOUR_API_KEY") as solver:
    print(f"Balance: ${solver.get_balance():.2f}")

    result = solver.solve_recaptcha_v2(RecaptchaV2Params(
        sitekey="6Le-SITEKEY",
        page_url="https://example.com"
    ))
    print(f"Solved in {result.solve_time:.1f}s")
    print(f"Token: {result.token[:50]}...")

Error handling

from captchaai_client import CaptchaAI, RecaptchaV2Params
from captchaai_client.exceptions import ZeroBalanceError, TaskTimeoutError

with CaptchaAI(api_key="YOUR_API_KEY", timeout=90) as solver:
    try:
        result = solver.solve_recaptcha_v2(RecaptchaV2Params(
            sitekey="6Le-SITEKEY",
            page_url="https://example.com"
        ))
    except ZeroBalanceError:
        print("Add funds to your CaptchaAI account")
    except TaskTimeoutError:
        print("Solve timed out — try again or check your parameters")

Reporting results

# Report incorrect solve for refund
solver.report_bad(result.task_id)

# Report correct solve (improves routing)
solver.report_good(result.task_id)

Troubleshooting

Problem	Cause	Fix
`WrongAPIKeyError`	Invalid or deactivated key	Verify key at captchaai.com dashboard
`TaskTimeoutError`	Solve taking too long	Increase `timeout` parameter or check sitekey
`ConnectionError`	Network issue	The wrapper retries on `ERROR_NO_SLOT_AVAILABLE` but not on network errors — add your own retry for connection failures

FAQ

Should I create one client per request or reuse it?

Reuse. The client uses requests.Session internally for connection pooling. Create one instance and pass it around.

Can I use this wrapper with async code?

This wrapper is synchronous. For async, replace requests with aiohttp and time.sleep with asyncio.sleep. See the Python asyncio + CaptchaAI guide.