Cliente CaptchaAI Python com validação Pydantic

Um sitekey vazio ou um pageurl sem https:// só aparecem como erro depois que a requisição já foi e voltou da API da CaptchaAI — normalmente uns 5 segundos perdidos por chamada, multiplicados por cada worker do pipeline que repete o mesmo parâmetro quebrado. A saída mais simples é validar antes de sair da máquina local: com Pydantic v2, um sitekey curto demais ou uma URL sem esquema disparam um ValidationError na hora, sem gastar uma única requisição HTTP.

Este guia monta um cliente Python completo para a API da CaptchaAI. Você vai sair daqui com:

• modelos Pydantic de requisição para reCAPTCHA v2, reCAPTCHA v3, Cloudflare Turnstile e CAPTCHA de imagem;
• modelos de resposta que transformam o JSON da API em objetos tipados, sem dict["key"] solto pelo código;
• uma classe CaptchaAI completa, com submissão, polling e tratamento de erro;
• uma tabela de erros comuns e cinco perguntas frequentes sobre validação, migração de versão e uso assíncrono.

Por que validar antes de chamar a API

Em um pipeline de QA rodando em sa-east-1 (São Paulo), por exemplo, um erro de parâmetro só é descoberto no fim da fila de polling — o worker já reservou uma thread do seu plano CaptchaAI, esperou o poll_interval e só então recebeu ERROR_WRONG_CAPTCHA_ID. Validar o payload antes do POST corta esse ciclo pela raiz.

Efeito colateral útil: como a validação acontece antes de qualquer chamada de rede, um sitekey ou pageurl malformado nunca chega a ser enviado para fora da sua máquina — o que também ajuda equipes que seguem a LGPD e tratam logs de teste como dado a ser minimizado.

O ganho na prática, comparando os dois cenários:

• Sem Pydantic: sitekey vazio só vira erro depois de uma chamada de API completa (uns 5 s de espera); parsing de resposta via dict["key"] quebra com KeyError; o editor não sugere nada sobre os parâmetros esperados.
• Com Pydantic: o mesmo sitekey vazio dispara ValidationError na hora, antes do requests.post; a resposta vira um modelo tipado com validação embutida; o IDE autocompleta cada campo.

Modelos Pydantic por tipo de CAPTCHA

Cada tipo suportado pela CaptchaAI ganha seu próprio modelo de requisição:

• RecaptchaV2Request — sitekey, pageurl e a flag invisible;
• RecaptchaV3Request — adiciona action e o score_qa esperado;
• TurnstileRequest — sitekey, pageurl e o cdata opcional do Cloudflare Turnstile;
• ImageRequest — imagem em base64, com um field_validator que já remove o prefixo data: se ele vier junto.

Do outro lado, os modelos de resposta (SubmitResponse, PollResponse, SolveResult) fazem o caminho inverso: pegam o JSON solto que a API devolve e expõem propriedades já tipadas, então result.token chega pronto para uso em vez de um dict["request"] que pode nem existir.

# 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)
    score_qa: float = Field(default=0.3, ge=0.1, le=0.9)

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

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")

Cliente que valida antes de enviar

A classe CaptchaAI reúne submissão (_submit), polling (_poll) e conversão para SolveResult em um único ponto — cada método público (solve_recaptcha_v2, solve_turnstile, solve_image...) recebe os parâmetros crus, instancia o modelo Pydantic correspondente e só então monta os parâmetros da requisição com to_params(). Se a validação falhar, o ValidationError sobe direto para quem chamou, antes de qualquer requests.post para in.php. O polling em _poll usa time.monotonic() para respeitar o timeout configurado e evita ficar preso indefinidamente em uma tarefa que nunca fica pronta.

# 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)

Exemplo: requisição válida e três formas de falhar cedo

O trecho abaixo usa staging.example.com como URL de teste — troque pelo domínio do seu próprio ambiente de QA. Repare que os três try/except capturam falhas em pontos diferentes: sitekey inválido nunca sai da máquina local, score_qa fora do intervalo também é barrado antes do envio, e só o erro de CaptchaAIError (saldo, chave errada etc.) depende de uma resposta real da API.

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",
        score_qa=1.5,  # Invalid — max is 0.9
    )
except ValidationError as e:
    print(e)
    # score_qa: Input should be less than or equal to 0.9

# 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}")

Dependências do projeto:

pip install pydantic requests

Erros comuns e como corrigir

A maioria dos erros abaixo aparece na primeira execução, quando os dados ainda não seguem exatamente o formato esperado pelos modelos.

Dica: rode o bloco de exemplo da seção anterior com valores propositalmente inválidos — sitekey vazio, score_qa fora do intervalo — para ver cada ValidationError na prática antes de apontar o cliente para dados reais.

Perguntas frequentes

A validação com Pydantic deixa a chamada mais lenta?

Não de forma perceptível. A validação roda em microssegundos; uma viagem de ida e volta até a API da CaptchaAI leva segundos. Barrar um parâmetro inválido antes do POST economiza muito mais tempo do que a validação custa.

Dá para reaproveitar esses modelos em outros tipos de CAPTCHA além dos que já estão no exemplo?

Sim. Crie uma nova subclasse de BaseModel com os campos que o tipo exige e um método to_params() que devolve o dicionário de parâmetros. Depois, adicione um método solve_* correspondente na classe CaptchaAI que instancia esse modelo e chama _solve.

Consigo usar esse cliente dentro de uma API própria feita com FastAPI?

Sim, e o encaixe é natural: os mesmos modelos Pydantic usados aqui podem servir como corpo de requisição (request body) de um endpoint FastAPI, sem duplicar a validação. A parte de submissão e polling continua igual — só muda quem chama client.solve_recaptcha_v2().

O que muda se eu migrar do Pydantic v1 para o v2?

A sintaxe de validadores muda: @validator vira @field_validator com @classmethod, e .dict() vira .model_dump(). Os exemplos deste artigo já usam a API do Pydantic v2 — se seu projeto ainda está no v1, use-os como referência para a migração.

Preciso validar a chave de API antes de cada chamada?

O construtor de CaptchaAI já faz uma checagem mínima (comprimento da chave) na inicialização do cliente, uma vez por processo. Erros de chave inválida ou saldo zerado — que só o servidor sabe informar — continuam chegando como CaptchaAIError na primeira chamada real.

Artigos relacionados

• Guia completo de Playwright com Python e CaptchaAI
• Como montar pipelines de CAPTCHA do lado do cliente com a CaptchaAI
• Validação de segurança de callback em webhooks da CaptchaAI

Próximos passos

Monte seu cliente CaptchaAI validado — pegue sua chave de API e comece a adicionar os modelos Pydantic ao seu projeto.

Guias relacionados:

• Construindo um wrapper Python para API CaptchaAI
• Referência de códigos de erro CaptchaAI
• API JSON CaptchaAI versus API de formulário