Crie um microsserviço de solução CAPTCHA com FastAPI e CaptchaAI

Quando vários serviços ou membros da equipe precisam de solução CAPTCHA, centralizá-los em um microsserviço evita a duplicação da lógica entre os projetos. O suporte assíncrono do FastAPI o torna uma solução perfeita - a resolução de CAPTCHA envolve esperar por respostas externas da API, que o assíncrono trata com eficiência sem bloquear threads.

Este guia cria um microsserviço FastAPI que aceita solicitações de resolução CAPTCHA sobre REST e retorna tokens resolvidos via CaptchaAI.

O que você precisa

Instale dependências:

pip install fastapi uvicorn httpx

Estrutura do projeto

captcha-service/
├── main.py          # FastAPI app with endpoints
├── solver.py        # CaptchaAI solving logic
└── requirements.txt

Módulo solucionador CaptchaAI

# solver.py
import httpx
import asyncio

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"


async def submit_task(params: dict) -> str:
    """Submit a CAPTCHA task and return the task ID."""
    params["key"] = API_KEY
    params["json"] = 1

    async with httpx.AsyncClient() as client:
        response = await client.post(f"{BASE_URL}/in.php", data=params)
        data = response.json()

    if data.get("status") != 1:
        raise ValueError(f"Submit error: {data.get('request')}")
    return data["request"]


async def poll_result(task_id: str, initial_wait: int = 15, max_attempts: int = 30) -> dict:
    """Poll for the CAPTCHA result."""
    await asyncio.sleep(initial_wait)

    async with httpx.AsyncClient() as client:
        for _ in range(max_attempts):
            response = await client.get(f"{BASE_URL}/res.php", params={
                "key": API_KEY, "action": "get", "id": task_id, "json": 1
            })
            data = response.json()

            if data.get("status") == 1:
                return {
                    "token": data["request"],
                    "user_agent": data.get("user_agent", "")
                }
            if data.get("request") != "CAPCHA_NOT_READY":
                raise ValueError(f"Solve error: {data['request']}")

            await asyncio.sleep(5)

    raise TimeoutError("Solve timed out")


async def solve_recaptcha_v2(sitekey: str, pageurl: str, enterprise: bool = False) -> dict:
    params = {"method": "userrecaptcha", "googlekey": sitekey, "pageurl": pageurl}
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)


async def solve_recaptcha_v3(sitekey: str, pageurl: str, action: str, enterprise: bool = False) -> dict:
    params = {
        "method": "userrecaptcha", "version": "v3",
        "googlekey": sitekey, "pageurl": pageurl, "action": action
    }
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)


async def solve_turnstile(sitekey: str, pageurl: str) -> dict:
    task_id = await submit_task({"method": "turnstile", "sitekey": sitekey, "pageurl": pageurl})
    return await poll_result(task_id, initial_wait=10)


async def solve_image(image_base64: str) -> dict:
    task_id = await submit_task({"method": "base64", "body": image_base64})
    return await poll_result(task_id, initial_wait=5, max_attempts=15)

Aplicativo FastAPI

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import solver

app = FastAPI(title="CaptchaAI Solver Service")


class RecaptchaV2Request(BaseModel):
    sitekey: str
    pageurl: str
    enterprise: bool = False


class RecaptchaV3Request(BaseModel):
    sitekey: str
    pageurl: str
    action: str
    enterprise: bool = False


class TurnstileRequest(BaseModel):
    sitekey: str
    pageurl: str


class ImageRequest(BaseModel):
    image_base64: str


class SolveResponse(BaseModel):
    token: str
    user_agent: Optional[str] = ""


@app.post("/solve/recaptcha-v2", response_model=SolveResponse)
async def solve_recaptcha_v2(req: RecaptchaV2Request):
    try:
        result = await solver.solve_recaptcha_v2(req.sitekey, req.pageurl, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/recaptcha-v3", response_model=SolveResponse)
async def solve_recaptcha_v3(req: RecaptchaV3Request):
    try:
        result = await solver.solve_recaptcha_v3(req.sitekey, req.pageurl, req.action, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/turnstile", response_model=SolveResponse)
async def solve_turnstile(req: TurnstileRequest):
    try:
        result = await solver.solve_turnstile(req.sitekey, req.pageurl)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/image", response_model=SolveResponse)
async def solve_image(req: ImageRequest):
    try:
        result = await solver.solve_image(req.image_base64)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.get("/health")
async def health():
    return {"status": "ok"}

Execute o serviço

uvicorn main:app --host 0.0.0.0 --port 8000

Exemplos de uso

Resolva reCAPTCHA v2

curl -X POST http://localhost:8000/solve/recaptcha-v2 \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkS...", "pageurl": "https://staging.example.com/qa-login"}'

Resolva Cloudflare Turnstile

curl -X POST http://localhost:8000/solve/turnstile \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "0x4AAAA...", "pageurl": "https://example.com/form"}'

Resposta:

{
  "token": "03AGdBq24PBCqLmOx2V4...",
  "user_agent": "Mozilla/5.0..."
}

Notas de proteção de implantação

• Leia a chave de API do ambiente e falhe rapidamente durante a inicialização se o segredo estiver faltando.
• Separe a validação da solicitação da execução do solucionador para que cargas inválidas nunca alcancem o caminho da chamada de saída.
• Retorne respostas de erro estruturadas que distinguem falhas de validação, falhas de solucionador e rejeição upstream.

Solução de problemas

Perguntas frequentes

Por que usar FastAPI para um serviço de resolução de CAPTCHA?

FastAPI lida com I/O assíncrono nativamente, o que é ideal para resolução de CAPTCHA, onde a maior parte do tempo é gasta aguardando a resposta de CaptchaAI. Várias solicitações podem ser processadas simultaneamente sem threading.

Posso adicionar autenticação aos endpoints?

Sim. Adicione injeção de dependência FastAPI com validação de cabeçalho de chave de API ou OAuth2 para restringir o acesso.

Quantas solicitações simultâneas isso pode lidar?

Limitado pelo limite de tarefas simultâneas do seu plano CaptchaAI. O próprio FastAPI pode lidar com milhares de conexões simultâneas.

Devo encaixar isso?

Sim. Adicione um Dockerfile com FROM python:3.11-slim, instale dependências e exponha a porta 8000.

Posso adicionar limitação de taxa?

Sim. Use slowapi ou um proxy reverso (nginx, Traefik) para limitar solicitações por cliente.

Crie seu microsserviço de solução CAPTCHA

Obtenha sua chave API emcaptchaai.com. Centralize a resolução de CAPTCHA com um microsserviço FastAPI.

Guias relacionados

• Como resolver reCAPTCHA v2 usando API
• Como resolver Cloudflare Turnstile usando API
• CaptchaAI assíncrono com aiohttp
• Resolução paralela de CAPTCHA