Uma rota Flask que resolve CAPTCHA de forma síncrona trava a aplicação enquanto espera a resposta da API — e é aí que a maioria das integrações emperra. Este guia monta uma API de resolução de CAPTCHA com a CaptchaAI dentro do Flask sem esse gargalo: classe de serviço reutilizável, rotas prontas para reCAPTCHA v2 e Cloudflare Turnstile, resolução em segundo plano com threading e organização em Blueprint para quando o projeto crescer.
O Flask é uma escolha natural aqui: leve o bastante para virar um microsserviço dedicado de CAPTCHA ou um wrapper dentro de algo maior. O padrão de integração é sempre o mesmo — enviar a tarefa, consultar o resultado —, só muda como você expõe isso: rota síncrona, tarefa em segundo plano ou Blueprint.
Montando a base: projeto e classe de serviço
Antes da primeira rota, três coisas: dependências, arquivos organizados e uma classe que concentre a conversa com a CaptchaAI.
Instalando as dependências
Duas bastam: o próprio Flask e a biblioteca requests.
pip install flask requests
Como organizar os arquivos
Separe a lógica de chamada à CaptchaAI das rotas HTTP desde o primeiro commit — isso facilita testar o solucionador isoladamente e trocar de framework depois, se for o caso:
myapp/
├── app.py
├── config.py
├── services/
│ └── captcha_solver.py
└── templates/
└── form.html
A classe que fala com a CaptchaAI
A classe CaptchaSolver concentra toda a comunicação com a CaptchaAI: enviar a tarefa para in.php, consultar res.php a cada 5 segundos até o status virar 1 e devolver o token pronto para uso. As rotas Flask nunca falam diretamente com a API — elas só chamam os métodos desta classe:
# services/captcha_solver.py
import time
import requests
class CaptchaSolver:
"""CaptchaAI solver service for Flask applications."""
API_BASE = "https://ocr.captchaai.com"
def __init__(self, api_key):
self.api_key = api_key
def solve_recaptcha_v2(self, sitekey, page_url):
"""Solve reCAPTCHA v2."""
return self._submit_and_poll({
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": page_url,
})
def solve_turnstile(self, sitekey, page_url):
"""Solve Cloudflare Turnstile."""
return self._submit_and_poll({
"method": "turnstile",
"sitekey": sitekey,
"pageurl": page_url,
})
def solve_image(self, image_base64):
"""Solve image CAPTCHA."""
return self._submit_and_poll({
"method": "base64",
"body": image_base64,
})
def get_balance(self):
"""Check API balance."""
resp = requests.get(f"{self.API_BASE}/res.php", params={
"key": self.api_key,
"action": "getbalance",
"json": 1,
}, timeout=30)
return float(resp.json().get("request", 0))
def _submit_and_poll(self, params, timeout=120):
"""Submit and poll for result."""
submit_data = {"key": self.api_key, "json": 1, **params}
resp = requests.post(f"{self.API_BASE}/in.php", data=submit_data, timeout=30)
resp.raise_for_status()
data = resp.json()
if data.get("status") != 1:
raise CaptchaSolveError(f"Submit failed: {data.get('request')}")
task_id = data["request"]
start = time.time()
while time.time() - start < timeout:
time.sleep(5)
result = requests.get(f"{self.API_BASE}/res.php", params={
"key": self.api_key,
"action": "get",
"id": task_id,
"json": 1,
}, timeout=30).json()
if result.get("status") == 1:
return result["request"]
if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
raise CaptchaSolveError("CAPTCHA unsolvable")
raise CaptchaSolveError("Solve timed out")
class CaptchaSolveError(Exception):
pass
Primeira rota: resolvendo CAPTCHA pela API
Com o serviço pronto, a aplicação Flask fica pequena: cada rota recebe a sitekey e a URL da página, chama o método correspondente do solucionador e devolve o token em JSON. O endpoint /balance serve para conferir o saldo antes de rodar um lote grande de tarefas:
# app.py
from flask import Flask, request, jsonify
from services.captcha_solver import CaptchaSolver, CaptchaSolveError
app = Flask(__name__)
app.config["CAPTCHAAI_API_KEY"] = "YOUR_API_KEY"
solver = CaptchaSolver(app.config["CAPTCHAAI_API_KEY"])
@app.route("/solve/recaptcha", methods=["POST"])
def solve_recaptcha():
"""Solve reCAPTCHA v2 via API."""
data = request.get_json()
sitekey = data.get("sitekey")
page_url = data.get("url")
if not sitekey or not page_url:
return jsonify({"error": "sitekey and url required"}), 400
try:
token = solver.solve_recaptcha_v2(sitekey, page_url)
return jsonify({"token": token})
except CaptchaSolveError as e:
return jsonify({"error": str(e)}), 500
@app.route("/solve/turnstile", methods=["POST"])
def solve_turnstile():
"""Solve Cloudflare Turnstile via API."""
data = request.get_json()
sitekey = data.get("sitekey")
page_url = data.get("url")
if not sitekey or not page_url:
return jsonify({"error": "sitekey and url required"}), 400
try:
token = solver.solve_turnstile(sitekey, page_url)
return jsonify({"token": token})
except CaptchaSolveError as e:
return jsonify({"error": str(e)}), 500
@app.route("/balance", methods=["GET"])
def check_balance():
"""Check CaptchaAI balance."""
balance = solver.get_balance()
return jsonify({"balance": balance})
if __name__ == "__main__":
app.run(debug=True, port=5000)
Como testar com curl
Suba o servidor com python app.py e valide as três rotas com curl antes de plugar um cliente real:
# Solve reCAPTCHA
curl -X POST http://localhost:5000/solve/recaptcha \
-H "Content-Type: application/json" \
-d '{"sitekey": "6Le-wvkSAAAA...", "url": "https://staging.example.com/qa-login"}'
# Solve Turnstile
curl -X POST http://localhost:5000/solve/turnstile \
-H "Content-Type: application/json" \
-d '{"sitekey": "0x4AAAAAAAC3DHQ...", "url": "https://example.com/signup"}'
# Check balance
curl http://localhost:5000/balance
Protegendo formulários Flask com o Cloudflare Turnstile
O caso acima resolve CAPTCHA para automação. Este aqui é o oposto: proteger um formulário do seu próprio site com o Cloudflare Turnstile e validar o token no servidor antes de processar o envio. O widget do Turnstile roda no navegador do visitante, o token chega no campo cf-turnstile-response do formulário e o Flask confirma esse token direto com a Cloudflare — sem precisar da CaptchaAI neste caso, já que quem resolve o desafio é o próprio usuário:
# app.py
from flask import Flask, request, render_template, redirect, url_for, flash
import requests as http_requests
app = Flask(__name__)
app.secret_key = "your-secret-key"
app.config["TURNSTILE_SITE_KEY"] = "0x4AAAAAAAC3DHQhMMQ_Rxrg"
app.config["TURNSTILE_SECRET_KEY"] = "0x4AAAAAAAC3DHQhYYY_secret"
def verify_turnstile(token, remote_ip=None):
"""Verify Turnstile token with Cloudflare."""
data = {
"secret": app.config["TURNSTILE_SECRET_KEY"],
"response": token,
}
if remote_ip:
data["remoteip"] = remote_ip
resp = http_requests.post(
"https://challenges.cloudflare.com/turnstile/v0/siteverify",
data=data,
timeout=10,
)
return resp.json().get("success", False)
@app.route("/contact", methods=["GET", "POST"])
def contact():
if request.method == "POST":
turnstile_token = request.form.get("cf-turnstile-response")
if not turnstile_token:
flash("CAPTCHA required")
return redirect(url_for("contact"))
if not verify_turnstile(turnstile_token, request.remote_addr):
flash("CAPTCHA verification failed")
return redirect(url_for("contact"))
# Process the form
name = request.form.get("name")
email = request.form.get("email")
# ... save or email the data
flash("Message sent successfully")
return redirect(url_for("contact"))
return render_template("form.html",
turnstile_sitekey=app.config["TURNSTILE_SITE_KEY"])
<!-- templates/form.html -->
<!DOCTYPE html>
<html>
<body>
<form method="post">
<input name="name" placeholder="Name" required>
<input name="email" type="email" placeholder="Email" required>
<textarea name="message" placeholder="Message" required></textarea>
<div class="cf-turnstile" data-sitekey="{{ turnstile_sitekey }}"></div>
<button type="submit">Send</button>
</form>
<script src="https://challenges.cloudflare.com/turnstile/v0/api.js" async defer></script>
</body>
</html>
Escalando: threading em segundo plano e Blueprint
Duas mudanças resolvem os problemas que aparecem fora do protótipo: uma rota que trava por minutos, e rotas de CAPTCHA espalhadas pela aplicação.
Resolvendo em segundo plano sem travar o Flask
O Flask padrão é síncrono: enquanto uma requisição espera o token da CaptchaAI, o worker que a atende fica ocupado e não atende mais ninguém. Para um endpoint que às vezes leva mais de um minuto — caso comum com reCAPTCHA v2 —, isso derruba o throughput rapidinho. A saída mais simples, sem trocar de framework, é disparar a resolução em uma threading.Thread e deixar o cliente consultar o status depois.
Threads do Python x threads do plano CaptchaAI
Vale separar dois conceitos que só coincidem no nome. A threading.Thread abaixo é uma thread do sistema operacional, local ao processo Flask. Já as threads de um plano CaptchaAI — por exemplo, BASIC (US$ 15/mês, 5 threads) ou ADVANCE (US$ 90/mês, 50 threads) — são quantas tarefas a CaptchaAI processa em paralelo para a sua conta, não quantas threading.Thread você dispara. Rodar 50 tarefas de uma vez contra um plano com 5 threads só faz as 45 excedentes esperarem na fila do lado da CaptchaAI.
import uuid
import threading
from flask import Flask, request, jsonify
from services.captcha_solver import CaptchaSolver, CaptchaSolveError
app = Flask(__name__)
solver = CaptchaSolver("YOUR_API_KEY")
# In-memory task storage (use Redis in production)
tasks = {}
def solve_in_background(task_id, captcha_type, sitekey, page_url):
"""Background CAPTCHA solver."""
try:
if captcha_type == "recaptcha_v2":
token = solver.solve_recaptcha_v2(sitekey, page_url)
elif captcha_type == "turnstile":
token = solver.solve_turnstile(sitekey, page_url)
else:
raise ValueError(f"Unknown type: {captcha_type}")
tasks[task_id] = {"status": "solved", "token": token}
except CaptchaSolveError as e:
tasks[task_id] = {"status": "failed", "error": str(e)}
@app.route("/solve/async", methods=["POST"])
def solve_async():
"""Submit CAPTCHA for background solving."""
data = request.get_json()
captcha_type = data.get("type", "recaptcha_v2")
sitekey = data.get("sitekey")
page_url = data.get("url")
if not sitekey or not page_url:
return jsonify({"error": "sitekey and url required"}), 400
task_id = str(uuid.uuid4())
tasks[task_id] = {"status": "pending"}
thread = threading.Thread(
target=solve_in_background,
args=(task_id, captcha_type, sitekey, page_url),
)
thread.start()
return jsonify({"task_id": task_id}), 202
@app.route("/solve/status/<task_id>")
def solve_status(task_id):
"""Check solving status."""
task = tasks.get(task_id)
if not task:
return jsonify({"error": "Task not found"}), 404
return jsonify(task)
Acompanhando o status da tarefa
O cliente recebe um task_id na hora e faz polling em /solve/status/<task_id> até o campo status mudar de pending para solved ou failed:
# Submit async solve
curl -X POST http://localhost:5000/solve/async \
-H "Content-Type: application/json" \
-d '{"type": "turnstile", "sitekey": "0x4AAA...", "url": "https://example.com"}'
# Returns: {"task_id": "abc-123-..."}
# Check status
curl http://localhost:5000/solve/status/abc-123-...
# Returns: {"status": "pending"} or {"status": "solved", "token": "..."}
Organizando rotas com Blueprint
Em uma aplicação com dezenas de rotas, misturar as de CAPTCHA com o resto do sistema dificulta a manutenção. Um Blueprint isola tudo sob um prefixo próprio (/api/captcha) e mantém a lógica de resolução em um único lugar:
# blueprints/captcha.py
from flask import Blueprint, request, jsonify, current_app
from services.captcha_solver import CaptchaSolver, CaptchaSolveError
captcha_bp = Blueprint("captcha", __name__, url_prefix="/api/captcha")
def get_solver():
return CaptchaSolver(current_app.config["CAPTCHAAI_API_KEY"])
@captcha_bp.route("/solve", methods=["POST"])
def solve():
data = request.get_json()
captcha_type = data.get("type")
sitekey = data.get("sitekey")
url = data.get("url")
solver = get_solver()
try:
if captcha_type == "recaptcha_v2":
token = solver.solve_recaptcha_v2(sitekey, url)
elif captcha_type == "turnstile":
token = solver.solve_turnstile(sitekey, url)
elif captcha_type == "image":
image_b64 = data.get("image")
token = solver.solve_image(image_b64)
else:
return jsonify({"error": f"Unknown type: {captcha_type}"}), 400
return jsonify({"token": token})
except CaptchaSolveError as e:
return jsonify({"error": str(e)}), 500
@captcha_bp.route("/balance")
def balance():
solver = get_solver()
return jsonify({"balance": solver.get_balance()})
# app.py
from flask import Flask
from blueprints.captcha import captcha_bp
app = Flask(__name__)
app.config["CAPTCHAAI_API_KEY"] = "YOUR_API_KEY"
app.register_blueprint(captcha_bp)
Problemas comuns e perguntas frequentes
Sintomas comuns e como corrigir
| Sintoma | Causa | Correção |
|---|---|---|
| A requisição trava por mais de 2 minutos | Resolução síncrona bloqueia o Flask | Use threading ou um padrão assíncrono |
ConnectionError |
API da CaptchaAI inacessível | Verifique rede e firewall |
| Token retorna vazio | Falha ao interpretar o JSON de resposta | Confira o formato da resposta |
| Verificação do Turnstile falha | Chave secreta incorreta | Revise a TURNSTILE_SECRET_KEY |
| Uso de memória cresce nas tarefas em segundo plano | O dicionário de tarefas nunca é limpo | Adicione TTL e uma rotina de limpeza |
Quanto tempo leva para resolver um CAPTCHA por essa integração?
Entre 15 e 120 segundos, dependendo do tipo de desafio — o Turnstile costuma responder bem mais rápido que o reCAPTCHA v2. É essa variação que torna a resolução síncrona arriscada em produção e justifica a versão em segundo plano deste guia.
Preciso me preocupar com a LGPD ao logar sitekey e token resolvido?
Sim, se o log também guarda IP, sessão ou outro dado que identifique o usuário. Evite gravar o token completo em texto puro em produção e trate o task_id como identificador de auditoria — ele não expõe o desafio resolvido.
O Flask-Limiter é suficiente para não gastar créditos à toa?
Ajuda bastante. Coloque flask-limiter na frente das rotas /solve/* para limitar quantas tarefas um mesmo cliente envia por minuto — toda chamada repetida ainda consome uma thread do seu plano CaptchaAI.
Como evito que o Gunicorn derrube a requisição antes do CAPTCHA ser resolvido?
Aumente o timeout do worker WSGI (gunicorn --timeout 180) para cobrir o pior caso, ou migre a rota para o padrão assíncrono deste guia — a requisição original responde em milissegundos e o cliente só consulta o resultado depois.
Dá para trocar o dicionário em memória por Celery e Redis mais tarde?
Sim, é o caminho natural quando o volume cresce. A troca é direta: o task_id continua sendo a chave, só a origem do armazenamento muda de um dict para o backend do Celery.
Qual padrão usar em cada caso
Para um endpoint isolado ou um script de QA, a rota síncrona resolve. Para tráfego real, prefira o padrão em segundo plano com threading — evita travar o Flask sem exigir uma fila externa. E numa aplicação com dezenas de rotas, o Blueprint mantém a lógica de CAPTCHA isolada e fácil de testar. As três formas reaproveitam a mesma classe CaptchaSolver.
Pegue sua chave de API em CaptchaAI e comece pela rota síncrona — é a forma mais rápida de validar a integração antes de partir para produção.
