脚本每隔几秒调用一次 /res.php 才能拿到结果,量一大就是两个问题:请求数蹭蹭涨,结果还总慢半拍。CaptchaAI 的回调(webhook)机制能直接砍掉这一步——识别完成后 token 自动 POST 到你的服务器,不用主动去问。下面从提交请求、搭建接收端到加安全校验,一步步带你接入。
回调 vs 轮询:先看结论
轮询简单,但请求多、延迟高;回调一次提交加一次接收就能拿到结果,代价是要有一个公网可访问的接收地址。该选哪种,看场景:
| 维度 | 轮询 | 回调 |
|---|---|---|
| API 请求次数 | 多(1 次提交 + N 次轮询) | 2 次(提交 + 回调) |
| 延迟 | 受轮询间隔限制 | 接近零延迟 |
| 服务器负担 | 客户端持续发请求 | CaptchaAI 主动推送 |
| 实现复杂度 | 简单 | 需要公网可访问的接收端点 |
| 防火墙要求 | 只需出站权限 | 必须放行入站 POST |
| 适合场景 | 单次脚本、临时任务 | 生产环境、批量处理管道 |
按场景选
| 场景 | 推荐方案 |
|---|---|
| 一次性脚本,识别量很少 | 轮询 |
| 生产环境批量处理 | 回调 |
| Serverless(Lambda、云函数) | 轮询(没有常驻进程接收回调) |
| 部署在内网、没有公网 IP | 轮询,或先用 ngrok/frp 做内网穿透测试 |
| 微服务架构,本身就有网关 | 回调 |
举个例子:采集服务跑在国内云主机上,出口防火墙常只放行 HTTPS 出站,轮询更省心;换成生产级批量任务,开一个入站端口接回调,比发几千次轮询划算。
回调机制是怎么跑起来的
整个过程只有两步:提交时带上回调地址,CaptchaAI 识别完成后主动把结果 POST 过来,不需要你的脚本再去问一次。
Standard polling approach:
Submit ──▶ Wait ──▶ Poll ──▶ Poll ──▶ Poll ──▶ Result
(many requests, wasted time between polls)
Callback approach:
Submit (with callback URL) ──▶ ... CaptchaAI solves ...
│
Your server receives POST ◀───────────────┘
(one request, instant delivery)
接入前先看常见故障
公网可达性是最容易踩坑的地方:
- 收不到回调——服务器没有公网地址:本地用 ngrok/frp 内网穿透测试,正式环境部署到有公网 IP 的云主机。
- 回调地址被拒绝——CaptchaAI 访问不到该 URL:确认域名解析正常、端口对外开放,安全组放行入站 POST。
- 只收到部分结果——个别回调失败或超时:记录所有回调,保留
/res.php轮询兜底。 - 同一任务收到多次回调——网络重试导致重复投递:用任务 ID 去重,重复直接丢弃。
提交请求时加上 pingback 参数
要用回调,只需要在提交请求里多加一个 pingback 字段,值填你的接收地址:
import requests
API_KEY = "YOUR_API_KEY"
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": "SITE_KEY",
"pageurl": "https://example.com",
"pingback": "https://your-server.com/captcha-callback",
"json": 1,
})
result = resp.json()
task_id = result["request"]
print(f"Task submitted: {task_id}")
# No polling needed — result comes via webhook
各种验证码类型都是同一个套路
不管识别哪种验证码,加 pingback 参数的方式都一样,下面是几个常见类型的写法:
# reCAPTCHA v2
data = {
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": "SITE_KEY",
"pageurl": "https://example.com",
"pingback": "https://your-server.com/callback",
"json": 1,
}
# Turnstile
data = {
"key": API_KEY,
"method": "turnstile",
"sitekey": "SITE_KEY",
"pageurl": "https://example.com",
"pingback": "https://your-server.com/callback",
"json": 1,
}
# Image CAPTCHA
data = {
"key": API_KEY,
"method": "base64",
"body": base64_image,
"pingback": "https://your-server.com/callback",
"json": 1,
}
用 Flask 搭建一个回调接收端
from flask import Flask, request
import threading
import logging
app = Flask(__name__)
logger = logging.getLogger(__name__)
# Store results by task ID
results = {}
events = {}
@app.route("/captcha-callback", methods=["POST", "GET"])
def captcha_callback():
"""Receive solved CAPTCHA tokens from CaptchaAI."""
# CaptchaAI sends parameters as query string or form data
task_id = request.args.get("id") or request.form.get("id")
code = request.args.get("code") or request.form.get("code")
if not task_id or not code:
logger.warning("Callback missing id or code")
return "ERROR", 400
logger.info("Received result for task %s", task_id)
results[task_id] = code
# Notify waiting threads
event = events.get(task_id)
if event:
event.set()
return "OK"
def wait_for_result(task_id, timeout=120):
"""Wait for a callback result."""
event = threading.Event()
events[task_id] = event
if task_id in results:
return results.pop(task_id)
event.wait(timeout=timeout)
if task_id in results:
return results.pop(task_id)
raise TimeoutError(f"No callback received for task {task_id}")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
这个接收端做了两件事:把 CaptchaAI POST 过来的 id 和 code 存进内存字典,再用 threading.Event 唤醒正在等这个任务的调用方——不需要你自己再写一层轮询逻辑。
串联成一个完整流程
import requests
import threading
import time
API_KEY = "YOUR_API_KEY"
CALLBACK_URL = "https://your-server.com/captcha-callback"
def solve_with_callback(sitekey, pageurl):
"""Submit CAPTCHA and wait for callback."""
# Submit with callback URL
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": pageurl,
"pingback": CALLBACK_URL,
"json": 1,
})
result = resp.json()
if result.get("status") != 1:
raise RuntimeError(f"Submit failed: {result.get('request')}")
task_id = result["request"]
print(f"Task {task_id} submitted, waiting for callback...")
# Wait for callback on the receiver
token = wait_for_result(task_id, timeout=120)
print(f"Token received via callback: {token[:50]}...")
return token
给回调端点加一层校验,防止被冒充
公网地址谁都能 POST 数据过来,处理前先做两件事:
- 提交时把
task_id记下来。 - 回调进来后核对
id是否在列表里,不在就拒绝。
代码实现:
from flask import Flask, request, abort
app = Flask(__name__)
# Store submitted task IDs to validate callbacks
pending_tasks = set()
def submit_captcha(sitekey, pageurl):
"""Submit and track task ID."""
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": pageurl,
"pingback": CALLBACK_URL,
"json": 1,
})
task_id = resp.json()["request"]
pending_tasks.add(task_id)
return task_id
@app.route("/captcha-callback", methods=["POST", "GET"])
def secure_callback():
"""Validate callback before processing."""
task_id = request.args.get("id") or request.form.get("id")
# Reject unknown task IDs
if task_id not in pending_tasks:
abort(403)
code = request.args.get("code") or request.form.get("code")
pending_tasks.discard(task_id)
results[task_id] = code
return "OK"
常见问题
回调地址必须用 HTTPS 吗?
生产环境强烈建议用 HTTPS——HTTP 明文传输,token 有被截获的风险。测试阶段用 HTTP 图方便可以,上线前务必切换。
本地开发环境没有公网 IP,怎么测试回调?
用 ngrok(或者 frp、natapp 这类国内也能用的内网穿透工具)把本地端口映射成临时公网地址,把拿到的 URL 填进 pingback 参数即可,调试完再切回正式地址。
CaptchaAI 的回调请求是 GET 还是 POST?
接收端要用 methods=["POST", "GET"] 同时接受——参数可能以查询字符串或表单数据发送。统一从 id/code 字段取值即可,不用关心具体方法。
怎么确认收到的回调确实来自 CaptchaAI,而不是别人伪造的?
提交时把任务 ID 记下来,回调进来先核对 id 是否在待处理列表里,不在就拒绝——具体实现见上面“加一层校验”代码。
相关指南
少发一堆轮询请求——试试 CaptchaAI 回调,识别完成 token 秒到。