团队同时维护几个爬虫或自动化项目时，往往每个项目都各写了一套验证码识别代码。更省事的做法是收敛成一个独立服务，其他项目调用一个 HTTP 接口就能拿到 token。
FastAPI 的异步模型很适配这个场景：识别耗时大多花在等 CaptchaAI 返回结果，async/await 不会让线程空等。本文用 FastAPI 搭建这样一个微服务，对外提供 REST 接口，内部通过 CaptchaAI API 识别 reCAPTCHA v2/v3、Turnstile 和图片验证码。出海产品的海外页面常见 reCAPTCHA、Turnstile，国内站点则以 GeeTest（极验）为主，统一到一个微服务更省心。
环境准备
开工前确认好这三样：
- CaptchaAI API Key：在 CaptchaAI 官网 注册获取
- Python 3.9 及以上
- FastAPI + httpx：用于异步 HTTP 调用
安装依赖（国内环境可加清华镜像参数加速）：
pip install fastapi uvicorn httpx
项目结构
solver.py 管识别逻辑，main.py 管路由和校验：
captcha-service/
├── main.py # FastAPI app with endpoints
├── solver.py # CaptchaAI solving logic
└── requirements.txt
CaptchaAI 识别模块：solver.py
四个 solve_* 函数各自拼好参数，统一走 submit_task 提交、poll_result 轮询：
# 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)
FastAPI 服务：main.py
四个 POST 接口对应四种验证码类型，外加 /health 接口供探活：
# 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"}
启动服务
uvicorn main:app --host 0.0.0.0 --port 8000
本地调试直接跑，上线后可放进容器交给编排系统管理。
调用示例
识别 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"}'
识别 Cloudflare Turnstile
curl -X POST http://localhost:8000/solve/turnstile \
-H "Content-Type: application/json" \
-d '{"sitekey": "0x4AAAA...", "pageurl": "https://example.com/form"}'
响应示例：
{
"token": "03AGdBq24PBCqLmOx2V4...",
"user_agent": "Mozilla/5.0..."
}
生产环境加固建议
- API Key 从环境变量读取，缺失时启动阶段就快速失败。
- 请求校验和求解器执行分开处理。
- 返回结构化错误，区分校验失败、求解器异常和上游拒绝。
常见故障排查
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 502 响应 | CaptchaAI 返回了错误 | 查看 detail 字段里的具体错误信息 |
| 识别超时 | 验证码处理耗时超出预期 | 调大 max_attempts，或检查 CaptchaAI 服务状态 |
| 连接被拒绝 | 服务没有启动 | 确认 uvicorn 监听的端口和请求端口一致 |
| 响应慢 | 阻塞了事件循环 | 确保用的是 httpx.AsyncClient，而不是同步的 requests |
常见问题
微服务能同时处理多少个验证码请求？
瓶颈在 CaptchaAI 的线程数，不在 FastAPI：
- BASIC（$15/月，5 线程）：同时最多 5 个任务
- STANDARD（$30/月，15 线程）：同时最多 15 个任务
某个实例挂了，其他项目的识别会跟着断吗？
只部署一个实例会。建议多实例前面挂反向代理（nginx、Traefik），配合 /health 做健康检查，异常实例自动摘掉。
需要给 /solve/* 接口加鉴权吗？
需要，至少加一层 API Key 请求头校验，内部调用也不例外——用 FastAPI 依赖注入几行代码就能挂上。
怎么知道每次识别花了多长时间？
在 submit_task、poll_result 前后打点计时，连同 CaptchaAI 返回的状态写进结构化日志，方便定位是提交慢还是轮询慢。
现在就把验证码识别封装成微服务
在 CaptchaAI 官网 注册获取 API Key，把这套 FastAPI 微服务接入你现有的项目，验证码识别只写一遍。
