Cliente Python para CaptchaAI con validación Pydantic

¿Cuánto tiempo se pierde cuando la API rechaza una solicitud mal construida? Con validación en local, prácticamente cero. Un cliente Python con modelos Pydantic comprueba el sitekey, la pageurl y los campos obligatorios en tu propia máquina, antes de que la petición HTTP salga hacia la API de CaptchaAI. Así, un sitekey vacío o una URL sin https:// se convierten en un ValidationError claro y al instante, en lugar de un ERROR_WRONG_CAPTCHA_ID que solo descubres tras esperar la respuesta.

En esta guía montas ese cliente tipado paso a paso:

• Los modelos de entrada y respuesta, que describen cada tipo de CAPTCHA y el resultado que esperas.
• El cliente, que envía la tarea, sondea el resultado y te devuelve un token ya tipado.
• El manejo de errores, que separa los fallos de validación de los que solo aparecen al llamar a la API.

Por qué validar con Pydantic antes de llamar a la API

Sin validación, cada parámetro incorrecto se paga con un round-trip completo: envías la tarea, esperas y solo entonces recibes el error. Con Pydantic inviertes el orden — el fallo salta antes de tocar la red y dice qué campo está mal.

Un modelo comprueba en tu propia máquina, antes de cualquier llamada:

• Que el sitekey tenga una longitud plausible y sin espacios sobrantes.
• Que la pageurl sea una URL válida y con esquema https://.
• Que los campos obligatorios de cada tipo de CAPTCHA estén presentes y bien tipados.

Piensa en una agencia que hace QA de flujos de checkout para varios marketplaces regionales, con un pipeline que resuelve cientos de reCAPTCHA v2 al día contra entornos de staging. Si un sitekey llega vacío desde una configuración mal cargada, cada intento pierde un round-trip y varios segundos de espera; en un lote grande, es tiempo de ejecución tirado. Validar en local con Pydantic reserva tus threads —por ejemplo, los 5 del plan BASIC ($15/mes)— para resoluciones reales.

Modelos de solicitud y respuesta

El archivo models.py reúne dos familias de modelos: los de solicitud validan lo que envías; los de respuesta parsean lo que devuelve la API.

• RecaptchaV2Request, RecaptchaV3Request, TurnstileRequest e ImageRequest para las entradas.
• SubmitResponse, PollResponse y SolveResult para las respuestas y el resultado final.

# models.py
from pydantic import BaseModel, Field, field_validator, HttpUrl
from enum import Enum
from typing import Optional

class CaptchaMethod(str, Enum):
    RECAPTCHA_V2 = "userrecaptcha"
    RECAPTCHA_V3 = "userrecaptcha"  # Differentiated by version field
    TURNSTILE = "turnstile"
    HCAPTCHA = "hcaptcha"
    IMAGE = "base64"
    GEETEST = "geetest"

class RecaptchaV2Request(BaseModel):
    """Parameters for solving reCAPTCHA v2."""
    sitekey: str = Field(min_length=20, max_length=100, description="Site's reCAPTCHA sitekey")
    pageurl: HttpUrl = Field(description="URL where CAPTCHA appears")
    invisible: bool = False
    cookies: Optional[str] = None

    @field_validator("sitekey")
    @classmethod
    def validate_sitekey(cls, v: str) -> str:
        if v.strip() != v:
            raise ValueError("Sitekey must not have leading/trailing whitespace")
        return v

    def to_params(self) -> dict:
        params = {
            "method": "userrecaptcha",
            "googlekey": self.sitekey,
            "pageurl": str(self.pageurl),
        }
        if self.invisible:
            params["invisible"] = "1"
        if self.cookies:
            params["cookies"] = self.cookies
        return params

class RecaptchaV3Request(BaseModel):
    """Parameters for solving reCAPTCHA v3."""
    sitekey: str = Field(min_length=20, max_length=100)
    pageurl: HttpUrl
    action: str = Field(default="verify", min_length=1, max_length=100)

    def to_params(self) -> dict:
        return {
            "method": "userrecaptcha",
            "version": "v3",
            "googlekey": self.sitekey,
            "pageurl": str(self.pageurl),
            "action": self.action,
        }

class TurnstileRequest(BaseModel):
    """Parameters for solving Cloudflare Turnstile."""
    sitekey: str = Field(min_length=10, max_length=100)
    pageurl: HttpUrl
    action: Optional[str] = None
    cdata: Optional[str] = None

    def to_params(self) -> dict:
        params = {
            "method": "turnstile",
            "sitekey": self.sitekey,
            "pageurl": str(self.pageurl),
        }
        if self.action:
            params["action"] = self.action
        if self.cdata:
            params["data"] = self.cdata
        return params

class ImageRequest(BaseModel):
    """Parameters for solving image/text CAPTCHA."""
    base64_image: str = Field(min_length=100, description="Base64-encoded image")
    case_sensitive: bool = False
    min_length: Optional[int] = Field(default=None, ge=1, le=50)
    max_length: Optional[int] = Field(default=None, ge=1, le=50)

    @field_validator("base64_image")
    @classmethod
    def validate_base64(cls, v: str) -> str:
        # Strip data URI prefix if present
        if v.startswith("data:"):
            parts = v.split(",", 1)
            if len(parts) == 2:
                return parts[1]
        return v

    def to_params(self) -> dict:
        params = {
            "method": "base64",
            "body": self.base64_image,
        }
        if self.case_sensitive:
            params["regsense"] = "1"
        if self.min_length is not None:
            params["min_len"] = str(self.min_length)
        if self.max_length is not None:
            params["max_len"] = str(self.max_length)
        return params

class SubmitResponse(BaseModel):
    """Parsed API submit response."""
    status: int
    request: str

    @property
    def success(self) -> bool:
        return self.status == 1

    @property
    def task_id(self) -> str:
        if not self.success:
            raise ValueError(f"No task ID — submission failed: {self.request}")
        return self.request

class PollResponse(BaseModel):
    """Parsed API poll response."""
    status: int
    request: str

    @property
    def ready(self) -> bool:
        return self.request != "CAPCHA_NOT_READY"

    @property
    def success(self) -> bool:
        return self.status == 1

    @property
    def token(self) -> str:
        if not self.success:
            raise ValueError(f"No token — solve failed: {self.request}")
        return self.request

class SolveResult(BaseModel):
    """Result of a successful solve."""
    token: str
    task_id: str
    solve_time: float = Field(description="Solve time in seconds")

El cliente CaptchaAI

El cliente reúne los modelos en una clase con dos operaciones internas —_submit para enviar la tarea y _poll para sondear el resultado— más un método por tipo de CAPTCHA. Solo llama a la red cuando la entrada ya es válida.

# client.py
import time
import requests
from pydantic import ValidationError

from models import (
    RecaptchaV2Request,
    RecaptchaV3Request,
    TurnstileRequest,
    ImageRequest,
    SubmitResponse,
    PollResponse,
    SolveResult,
)

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

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

class CaptchaAI:
    def __init__(self, api_key: str, poll_interval: int = 5, timeout: int = 180):
        if not api_key or len(api_key) < 10:
            raise ValueError("Invalid API key")
        self.api_key = api_key
        self.poll_interval = poll_interval
        self.timeout = timeout

    def _submit(self, params: dict) -> str:
        params["key"] = self.api_key
        params["json"] = 1

        resp = requests.post(SUBMIT_URL, data=params, timeout=30)
        result = SubmitResponse.model_validate(resp.json())

        if not result.success:
            raise CaptchaAIError(result.request, "Submit failed")

        return result.task_id

    def _poll(self, task_id: str) -> str:
        start = time.monotonic()

        while time.monotonic() - start < self.timeout:
            time.sleep(self.poll_interval)

            resp = requests.get(RESULT_URL, params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1,
            }, timeout=15)

            result = PollResponse.model_validate(resp.json())

            if not result.ready:
                continue

            if result.success:
                return result.token

            raise CaptchaAIError(result.request, "Solve failed")

        raise CaptchaAIError("TIMEOUT", f"Task {task_id} timed out after {self.timeout}s")

    def _solve(self, params: dict) -> SolveResult:
        start = time.monotonic()
        task_id = self._submit(params)
        token = self._poll(task_id)
        elapsed = time.monotonic() - start

        return SolveResult(
            token=token,
            task_id=task_id,
            solve_time=round(elapsed, 1),
        )

    def solve_recaptcha_v2(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
        """Solve reCAPTCHA v2 with validated parameters."""
        req = RecaptchaV2Request(sitekey=sitekey, pageurl=pageurl, **kwargs)
        return self._solve(req.to_params())

    def solve_recaptcha_v3(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
        """Solve reCAPTCHA v3 with validated parameters."""
        req = RecaptchaV3Request(sitekey=sitekey, pageurl=pageurl, **kwargs)
        return self._solve(req.to_params())

    def solve_turnstile(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
        """Solve Cloudflare Turnstile with validated parameters."""
        req = TurnstileRequest(sitekey=sitekey, pageurl=pageurl, **kwargs)
        return self._solve(req.to_params())

    def solve_image(self, base64_image: str, **kwargs) -> SolveResult:
        """Solve image/text CAPTCHA with validated parameters."""
        req = ImageRequest(base64_image=base64_image, **kwargs)
        return self._solve(req.to_params())

    def get_balance(self) -> float:
        """Get current account balance."""
        resp = requests.get(RESULT_URL, params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        }, timeout=10)
        result = SubmitResponse.model_validate(resp.json())
        return float(result.request)

El cliente expone un método por tipo de CAPTCHA:

• solve_recaptcha_v2 y solve_recaptcha_v3 para reCAPTCHA.
• solve_turnstile para Cloudflare Turnstile.
• solve_image para CAPTCHA de imagen o texto.

Cada método construye su modelo, lo valida y delega en _solve; si la entrada no cuadra, el ValidationError salta ahí mismo y la red ni se toca.

Cómo usar el cliente

Una solicitud válida pasa la validación y llega a la API; una inválida se corta en el ValidationError, sin gastar una llamada. Los fallos en tiempo de ejecución quedan capturados por CaptchaAIError. El ejemplo recorre cuatro casos:

• Una solicitud válida que supera la validación y resuelve el CAPTCHA.
• Un sitekey vacío que Pydantic rechaza al instante, sin llamada a la red.
• Una entrada inválida que se detiene antes de salir hacia la API.
• Un error devuelto por la API, atrapado por CaptchaAIError.

from pydantic import ValidationError
from client import CaptchaAI, CaptchaAIError

client = CaptchaAI("YOUR_API_KEY", timeout=120)

# Valid request — passes validation, calls API
result = client.solve_recaptcha_v2(
    sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
    pageurl="https://staging.example.com/qa-login",
)
print(f"Token: {result.token[:40]}...")
print(f"Solved in {result.solve_time}s")

# Invalid sitekey — caught immediately, no API call
try:
    client.solve_recaptcha_v2(sitekey="", pageurl="https://example.com")
except ValidationError as e:
    print(e)
    # sitekey: String should have at least 20 characters

# Invalid score — caught before API call
try:
    client.solve_recaptcha_v3(
        sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
        pageurl="https://example.com",
    )
except ValidationError as e:
    print(e)

# API error — caught during request
try:
    result = client.solve_turnstile(
        sitekey="0x4AAAAAAADnPIDROrmt1Wwj",
        pageurl="https://example.com",
    )
except CaptchaAIError as e:
    print(f"API error: {e.code}")

Instala las dependencias:

pip install pydantic requests

Solución de problemas

Casi todos los fallos caen en una de dos categorías: la validación local que rechaza la entrada, o el error que devuelve la API en tiempo de ejecución. Esta tabla reúne los más habituales:

Preguntas frecuentes

Estas son las dudas que suelen surgir al montar el cliente por primera vez.

¿Qué tipos de CAPTCHA puedo resolver con este cliente?

El cliente incluye modelos para reCAPTCHA v2 y v3, Cloudflare Turnstile e imagen/texto — todos tipos que CaptchaAI resuelve. Puedes añadir otros compatibles, como GeeTest v3, siguiendo el mismo patrón. El enum contiene una entrada hcaptcha, pero CaptchaAI no resuelve hCaptcha ni FunCaptcha, así que no crees un modelo de solicitud para ellos.

¿Funciona con Pydantic v1 o necesito la v2?

Necesitas Pydantic v2. El código usa field_validator y model_validate, APIs propias de la versión 2; en la v1 lanzará errores de importación. Instala la versión correcta con pip install 'pydantic>=2.0'.

¿La validación de Pydantic me protege de todos los errores de la API?

No. Pydantic valida la forma de la solicitud —longitud del sitekey, esquema de la pageurl, campos obligatorios— antes de la llamada. No puede saber si el sitekey corresponde a un sitio real, si tus threads están libres o si la resolución agotará el tiempo de espera. Por eso el cliente conserva CaptchaAIError para los fallos que solo aparecen en tiempo de ejecución.

¿Puedo usarlo de forma asíncrona con httpx?

Sí. Cambia requests por httpx.AsyncClient y convierte _submit, _poll y los métodos de resolución en async. Los modelos de Pydantic no cambian: validan de forma síncrona antes de la llamada HTTP asíncrona.

Artículos relacionados

• Pipeline de pruebas automatizadas con CaptchaAI
• Cómo construir pipelines de CAPTCHA en el cliente
• Validar la seguridad de webhooks con CaptchaAI

Crea tu cliente validado con CaptchaAI

Monta tu cliente tipado y deja de gastar peticiones en parámetros mal formados: obtén tu API key y añade los modelos de Pydantic.

Guías relacionadas:

• Referencia completa de códigos de error de CaptchaAI
• CaptchaAI JSON API vs Form API: qué formato elegir