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

Um único endpoint HTTP capaz de resolver reCAPTCHA v2, reCAPTCHA v3, Cloudflare Turnstile e CAPTCHA de imagem — sem duplicar a mesma lógica de chamada à API em cada projeto Python da equipe. É isso que este guia constrói: um microsserviço FastAPI que centraliza a resolução de CAPTCHA via CaptchaAI e expõe endpoints REST simples para qualquer serviço interno consumir.

Por que FastAPI, e não Flask puro? Porque resolver um CAPTCHA é, na prática, esperar: a CaptchaAI leva de segundos a mais de um minuto para devolver o token, dependendo do tipo. Nesse tempo, uma aplicação assíncrona atende centenas de outras requisições em vez de travar uma thread — exatamente o cenário para o qual o async/await do FastAPI foi desenhado.

Pré-requisitos para o microsserviço FastAPI e CaptchaAI

Antes de começar, você precisa de três coisas:

Instale as dependências:

pip install fastapi uvicorn httpx

Com isso instalado, o serviço fica pronto para subir em poucos minutos.

Estrutura do projeto do microsserviço

Organize os três arquivos assim — separar a lógica de chamada à CaptchaAI (solver.py) da camada HTTP (main.py) facilita testar cada parte isoladamente:

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

O módulo que resolve CAPTCHA com a API da CaptchaAI

O arquivo solver.py concentra toda a comunicação com a CaptchaAI: enviar a tarefa para in.php, aguardar e consultar o resultado em res.php até o token ficar pronto. Cada tipo de CAPTCHA (reCAPTCHA v2, reCAPTCHA v3, Turnstile, imagem) tem sua própria função de conveniência, mas todas reaproveitam submit_task e poll_result por baixo dos panos.

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

Repare no initial_wait de cada função: é o tempo de espera antes da primeira consulta, calibrado pela velocidade típica daquele tipo de CAPTCHA — imagem resolve bem mais rápido que reCAPTCHA v2, então não faz sentido esperar 20 segundos antes da primeira tentativa em solve_image.

Os endpoints REST do FastAPI

Com o módulo solucionador pronto, o main.py expõe cada tipo de CAPTCHA como um endpoint REST independente. Os modelos Pydantic validam o payload antes mesmo de a requisição chegar à lógica de resolução — o que separa uma falha de validação de uma falha real do solucionador.

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

Suba o serviço FastAPI localmente

Com os dois arquivos no lugar, suba o serviço:

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

O FastAPI em si lida com milhares de conexões simultâneas; o gargalo real quase sempre é o limite de threads do seu plano CaptchaAI, não o servidor Python.

Teste os endpoints com cURL

Valide os endpoints com cURL antes de plugar um cliente real:

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..."
}

Segurança e implantação em produção

Antes de expor esse microsserviço para outros times, vale reforçar alguns pontos:

• Leia a chave de API de uma variável de ambiente (ou secret manager) e falhe rápido na inicialização se ela estiver ausente.
• Mantenha a validação da requisição separada da execução do solucionador, como o Pydantic já faz acima: uma carga inválida nunca deve alcançar a chamada de saída para a CaptchaAI.
• Diferencie os erros na resposta: falha de validação (422), falha do solucionador (502, como no exemplo) e erro de configuração (500) — isso poupa horas de debug.
• Rodar perto de São Paulo (sa-east-1 na AWS) reduz a latência até a CaptchaAI em milissegundos, mas o gargalo real quase sempre é o tempo de resolução, não a rede.
• Padronize o deploy com um Dockerfile simples (FROM python:3.11-slim) e, para limitar requisições por cliente, use slowapi ou um proxy reverso (nginx, Traefik).
• Se o serviço loga qual token foi resolvido para qual pageurl, trate esse log como dado sensível sob a LGPD: evite gravar tokens completos ou URLs autenticadas em texto puro, com retenção curta.

Solução de problemas comuns

Se algo não funcionar como esperado, comece por aqui:

Perguntas frequentes

Reunimos as dúvidas mais comuns de quem já colocou esse microsserviço em produção.

Por que usar FastAPI em vez de Flask para esse serviço?

O FastAPI trata I/O assíncrono nativamente, e resolver CAPTCHA é basicamente esperar a resposta da CaptchaAI, não processar CPU. Com um framework síncrono, cada requisição em espera ocupa uma thread do servidor; com FastAPI e httpx.AsyncClient, centenas de tarefas ficam pendentes ao mesmo tempo sem esgotar o processo.

Como adiciono autenticação nos endpoints do microsserviço?

Use a injeção de dependência do FastAPI (Depends) com validação de um cabeçalho de chave própria do seu microsserviço — diferente da chave da CaptchaAI — ou integre OAuth2 se o serviço for consumido por vários times. Assim nenhum processo na rede interna chama o endpoint sem controle.

É necessário rodar o serviço perto da CaptchaAI para reduzir a latência?

Não é obrigatório, mas ajuda. Como o tempo de resolução domina o total — de alguns segundos a mais de um minuto, dependendo do tipo — a latência de rede costuma ser irrelevante: mesmo hospedando em sa-east-1 (São Paulo), a diferença é de milissegundos frente ao tempo de resolução.

Como lido com a LGPD ao registrar logs de tokens resolvidos?

Trate o token e a pageurl associada como dados potencialmente sensíveis: evite gravá-los em texto puro em logs persistentes, restrinja o acesso aos registros e defina uma retenção curta. É boa prática de qualquer forma, já que o token normalmente perde validade rapidamente no site de destino.

Quais tipos de CAPTCHA esse microsserviço resolve sem alterar o código?

Do jeito que o solver.py está, ele cobre quatro tipos:

• reCAPTCHA v2, reCAPTCHA v3, Cloudflare Turnstile e CAPTCHA de imagem — 4 dos 12 tipos com disponibilidade geral (GA) da CaptchaAI.
• Mais 3 tipos em beta, fora do escopo deste código: CaptchaFox (beta), Friendly Captcha (beta) e Lemin (beta).

Para adicionar outro tipo, replique o padrão de solve_turnstile: monte os parâmetros do method correspondente e reaproveite submit_task/poll_result.

Crie seu microsserviço de solução CAPTCHA

Pegue sua chave de API em captchaai.com e coloque esse microsserviço no ar em minutos.

Depois é só apontar os outros serviços da sua stack para os endpoints REST, em vez de duplicar a lógica de resolução em cada um deles.

Guias relacionados

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