Reusable CAPTCHA Modules for Client Projects

Every new client project means the same CAPTCHA-solving code written from scratch — submit, poll, parse, retry. Instead, package that logic into a module you import everywhere. This guide shows how.

What the module handles

A reusable module should encapsulate:

• Task submission — format parameters, POST to CaptchaAI
• Result polling — wait loop with configurable timeout
• Error handling — map error codes to actionable responses
• Configuration — API key, proxy, timeouts
• CAPTCHA type routing — one interface for all supported types

Python module

File structure

captcha_solver/
├── __init__.py
├── solver.py
├── config.py
└── exceptions.py

config.py

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class SolverConfig:
    api_key: str
    submit_url: str = "https://ocr.captchaai.com/in.php"
    result_url: str = "https://ocr.captchaai.com/res.php"
    poll_interval: int = 5
    max_wait: int = 120
    max_retries: int = 2
    proxy: Optional[str] = None
    proxytype: Optional[str] = None

exceptions.py

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

class ZeroBalanceError(CaptchaError):
    pass

class UnsolvableError(CaptchaError):
    pass

class TimeoutError(CaptchaError):
    pass

solver.py

import requests
import time
from typing import Optional
from .config import SolverConfig
from .exceptions import CaptchaError, ZeroBalanceError, UnsolvableError, TimeoutError

ERROR_MAP = {
    "ERROR_ZERO_BALANCE": ZeroBalanceError,
    "ERROR_CAPTCHA_UNSOLVABLE": UnsolvableError,
}

class CaptchaSolver:
    def __init__(self, config: SolverConfig):
        self.config = config

    def solve_recaptcha_v2(self, sitekey: str, page_url: str, **kwargs) -> str:
        return self._solve("userrecaptcha", {
            "googlekey": sitekey,
            "pageurl": page_url,
            **kwargs
        })

    def solve_turnstile(self, sitekey: str, page_url: str, **kwargs) -> str:
        return self._solve("turnstile", {
            "sitekey": sitekey,
            "pageurl": page_url,
            **kwargs
        })

    def solve_image(self, image_base64: str, **kwargs) -> str:
        return self._solve("base64", {
            "body": image_base64,
            **kwargs
        })

    def _solve(self, method: str, params: dict) -> str:
        for attempt in range(self.config.max_retries + 1):
            try:
                task_id = self._submit(method, params)
                return self._poll(task_id)
            except UnsolvableError:
                if attempt < self.config.max_retries:
                    continue
                raise
            except ZeroBalanceError:
                raise

    def _submit(self, method: str, params: dict) -> str:
        data = {
            "key": self.config.api_key,
            "method": method,
            "json": 1,
            **params
        }

        if self.config.proxy:
            data["proxy"] = self.config.proxy
            data["proxytype"] = self.config.proxytype

        resp = requests.post(self.config.submit_url, data=data, timeout=15)
        result = resp.json()

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

        error_code = result.get("error_text", result.get("request", "UNKNOWN"))
        error_class = ERROR_MAP.get(error_code, CaptchaError)
        raise error_class(error_code)

    def _poll(self, task_id: str) -> str:
        elapsed = 0
        while elapsed < self.config.max_wait:
            time.sleep(self.config.poll_interval)
            elapsed += self.config.poll_interval

            resp = requests.get(self.config.result_url, params={
                "key": self.config.api_key,
                "action": "get",
                "id": task_id,
                "json": 1
            }, timeout=10)
            result = resp.json()

            if result.get("status") == 1:
                return result["request"]
            elif result.get("request") == "CAPCHA_NOT_READY":
                continue
            else:
                error_code = result.get("error_text", result.get("request", "UNKNOWN"))
                error_class = ERROR_MAP.get(error_code, CaptchaError)
                raise error_class(error_code)

        raise TimeoutError("TIMEOUT", f"Task {task_id} timed out after {self.config.max_wait}s")

__init__.py

from .solver import CaptchaSolver
from .config import SolverConfig
from .exceptions import CaptchaError, ZeroBalanceError, UnsolvableError, TimeoutError

__all__ = [
    "CaptchaSolver",
    "SolverConfig",
    "CaptchaError",
    "ZeroBalanceError",
    "UnsolvableError",
    "TimeoutError",
]

Usage in client projects

from captcha_solver import CaptchaSolver, SolverConfig, ZeroBalanceError

config = SolverConfig(
    api_key="YOUR_API_KEY",
    proxy="host:port:user:pass",
    proxytype="HTTP",
    max_wait=90
)

solver = CaptchaSolver(config)

# reCAPTCHA v2
try:
    token = solver.solve_recaptcha_v2(
        sitekey="6Le-SITEKEY",
        page_url="https://target.com/form"
    )
    print(f"Token: {token[:40]}...")
except ZeroBalanceError:
    print("Account balance is zero — stop all tasks")
except Exception as e:
    print(f"Solve failed: {e}")

Node.js module

captcha-solver.js

const axios = require("axios");

class CaptchaSolver {
  constructor({ apiKey, proxy, proxytype, pollInterval = 5000, maxWait = 120000, maxRetries = 2 }) {
    this.apiKey = apiKey;
    this.proxy = proxy || null;
    this.proxytype = proxytype || null;
    this.pollInterval = pollInterval;
    this.maxWait = maxWait;
    this.maxRetries = maxRetries;
    this.submitUrl = "https://ocr.captchaai.com/in.php";
    this.resultUrl = "https://ocr.captchaai.com/res.php";
  }

  async solveRecaptchaV2(sitekey, pageUrl, extra = {}) {
    return this._solve("userrecaptcha", { googlekey: sitekey, pageurl: pageUrl, ...extra });
  }

  async solveTurnstile(sitekey, pageUrl, extra = {}) {
    return this._solve("turnstile", { sitekey, pageurl: pageUrl, ...extra });
  }

  async solveImage(imageBase64, extra = {}) {
    return this._solve("base64", { body: imageBase64, ...extra });
  }

  async _solve(method, params) {
    let lastError;
    for (let attempt = 0; attempt <= this.maxRetries; attempt++) {
      try {
        const taskId = await this._submit(method, params);
        return await this._poll(taskId);
      } catch (err) {
        lastError = err;
        if (err.code === "ERROR_ZERO_BALANCE") throw err;
        if (err.code !== "ERROR_CAPTCHA_UNSOLVABLE") throw err;
      }
    }
    throw lastError;
  }

  async _submit(method, params) {
    const data = { key: this.apiKey, method, json: 1, ...params };
    if (this.proxy) {
      data.proxy = this.proxy;
      data.proxytype = this.proxytype;
    }

    const resp = await axios.post(this.submitUrl, null, { params: data, timeout: 15000 });

    if (resp.data.status === 1) return resp.data.request;

    const code = resp.data.error_text || resp.data.request;
    const err = new Error(code);
    err.code = code;
    throw err;
  }

  async _poll(taskId) {
    let elapsed = 0;

    while (elapsed < this.maxWait) {
      await new Promise((r) => setTimeout(r, this.pollInterval));
      elapsed += this.pollInterval;

      const resp = await axios.get(this.resultUrl, {
        params: { key: this.apiKey, action: "get", id: taskId, json: 1 },
        timeout: 10000,
      });

      if (resp.data.status === 1) return resp.data.request;
      if (resp.data.request === "CAPCHA_NOT_READY") continue;

      const code = resp.data.error_text || resp.data.request;
      const err = new Error(code);
      err.code = code;
      throw err;
    }

    const err = new Error(`Timeout waiting for task ${taskId}`);
    err.code = "TIMEOUT";
    throw err;
  }
}

module.exports = { CaptchaSolver };

Usage

const { CaptchaSolver } = require("./captcha-solver");

const solver = new CaptchaSolver({
  apiKey: "YOUR_API_KEY",
  proxy: "host:port:user:pass",
  proxytype: "HTTP",
});

(async () => {
  try {
    const token = await solver.solveTurnstile(
      "0x4AAAA-SITEKEY",
      "https://target.com/login"
    );
    console.log(`Token: ${token.slice(0, 40)}...`);
  } catch (err) {
    console.error(`Failed: ${err.code} — ${err.message}`);
  }
})();

Configuration per client

Override defaults per client without changing the module:

# Client A — fast timeout, residential proxy
client_a = CaptchaSolver(SolverConfig(
    api_key="YOUR_API_KEY",
    proxy="res-proxy:port:user:pass",
    proxytype="HTTP",
    max_wait=60
))

# Client B — no proxy, longer timeout
client_b = CaptchaSolver(SolverConfig(
    api_key="YOUR_API_KEY",
    max_wait=180
))

Troubleshooting

FAQ

Should I publish the module to PyPI or npm?

Only if you plan to share across teams. For internal use, a private Git repo installed via pip install git+... or npm install git+... is simpler.

Can I add async support?

Yes. Replace requests with aiohttp in Python, or the module already uses async in Node.js. The interface stays the same.

How do I handle different API keys per client?

Create a separate SolverConfig (or constructor options in Node.js) per client, each with its own api_key.

Build reusable modules with CaptchaAI

Start building modular CAPTCHA solutions at captchaai.com.

Related guides

• Building Client CAPTCHA Pipelines
• Implementing Retry Logic
• Proxy Configuration Guide
• Error Codes Reference