使用 aiohttp 求解并提交 reCAPTCHA
使用 aiohttp 直接调用 REST API：创建一个任务，轮询直到就绪，然后向目标提交 solution.token。将 API 请求和目标请求放在独立的会话中，确保工作区 Bearer 令牌绝不会发送给目标。
此示例需要 Python 3.11+，并将创建、轮询和向目标提交全部限制在同一个 180 秒期限内。这是应用超时设置，并非承诺的解题时间。
本页内容
设置站点、代理和目标请求
- 将 SOLVETHREE_API_KEY 设为工作区 API 密钥。请将其保留在服务器端。
- 将 SOLVETHREE_PROXY 设为服务解题时应使用的代理 URL。向目标提交时使用同一个代理。
- 将 accounts.example.com 和 login 操作替换为集成所用的主机名和操作名称。该站点需要有一个当前工作区可用的有效配置；请在创建任务之前设置配置。
- 将目标登录 URL 和表单数据替换为实际的应用请求。示例只发送 g-recaptcha-response；请添加目标所需的其他字段，并使用目标预期的令牌字段。
任务发送 site、action 和一个 proxy。默认使用 reCAPTCHA Enterprise v3；不要添加 type 或 url 字段。如果目标流程需要先前请求中的 Cookie，请通过目标会话发出这些请求——示例从全新会话开始。
创建一次，轮询，然后提交
# pip install aiohttp
import asyncio
import os
import uuid
from urllib.parse import urlsplit, unquote
import aiohttp
async def main():
api = "https://api.solvethree.com/v1"
proxy = os.environ["SOLVETHREE_PROXY"]
task = {"site": "accounts.example.com", "action": "login", "proxy": proxy, "requestId": str(uuid.uuid4())}
task_id = None
try:
async with asyncio.timeout(180):
async with aiohttp.ClientSession(headers={"Authorization": "Bearer " + os.environ["SOLVETHREE_API_KEY"]}) as client:
async with client.post(api + "/tasks", params={"waitMs": 0}, json=task, allow_redirects=False) as response:
response.raise_for_status()
if response.status >= 300: raise RuntimeError("Unexpected API redirect")
result = await response.json()
task_id = result["id"]
while result["status"] in ("pending", "solving"):
async with client.get(api + "/tasks/" + task_id, params={"waitMs": 15000}, allow_redirects=False) as response:
response.raise_for_status()
if response.status >= 300: raise RuntimeError("Unexpected API redirect")
result = await response.json()
if result["status"] in ("pending", "solving"): await asyncio.sleep(1)
if result["status"] != "ready":
raise RuntimeError("Task failed: " + str(result.get("errorCode", "unknown")))
solution = result["solution"]
# A separate session prevents the API Bearer header reaching the target.
proxy_url = urlsplit(proxy)
netloc = proxy_url.netloc.rsplit("@", 1)[-1]
server = f"{proxy_url.scheme}://{netloc}"
auth = aiohttp.BasicAuth(unquote(proxy_url.username), unquote(proxy_url.password or "")) if proxy_url.username else None
async with aiohttp.ClientSession() as target:
async with target.post("https://accounts.example.com/login", proxy=server, proxy_auth=auth,
headers={"User-Agent": solution["userAgent"], "sec-ch-ua": solution["secChUa"]},
data={"g-recaptcha-response": solution["token"]}, allow_redirects=False) as response:
print("Application HTTP status:", response.status)
except TimeoutError:
print("Timed out; recover requestId", task["requestId"], "taskId", task_id)
raise
asyncio.run(main())这里直接使用异步 REST，而不是官方 Python SDK 的异步方法。它向 /v1/tasks 发送一次创建任务的 POST 请求，保存返回的 id，然后使用 GET 请求轮询该任务。pending 和 solving 状态会继续循环；ready 状态允许提交；任务进入 failed 状态时停止循环。
任务请求体中的 proxy 告诉 solvethree 应使用哪个代理。aiohttp 的 proxy 和 proxy_auth 参数仅用于目标 HTTP 调用，不用于 API 调用。示例将代理凭据与代理服务器 URL 分开，并使用返回的 userAgent 和 secChUa 提交令牌。
会话被刻意分开：只有 API 会话包含 Authorization 请求头。重定向已禁用，API 重定向会被视为错误，而不会被跟随。
- API 请求返回 HTTP 错误。
示例在读取成功响应内容之前检查 HTTP 状态。
试试这样做： 将 HTTP 错误与任务失败分开处理。重试之前，请修正无效输入或授权问题。
- 任务进入 failed 状态。
检查任务响应中的 errorCode 和 error。示例抛出异常时仅包含 errorCode。
试试这样做： 停止轮询该任务，并处理报告的原因。修复所报告的问题后再重试。
- 任务已就绪，但应用拒绝了请求。
检查发出的令牌字段、目标 URL、返回的浏览器身份和代理，以及所需的表单字段或 Cookie。
试试这样做： 通过同一个代理，使用返回的身份提交 solution.token。根据目标的成功判定标准解读应用响应——仅凭打印出的 HTTP 状态无法确定请求已被接受。
超时后恢复，无需再次付费解题
创建请求超时并不能证明创建失败。示例将原始输入保留在 task 中，并在期限到期时打印其 requestId 和 task_id。这些变量仅在进程运行期间存在；打印出的 ID 并不是完整的恢复记录。
- 1.
创建前保存原始请求
要在重启后恢复，请安全保存完整的任务请求体，包括其中的 requestId 和 proxy。示例每次运行都会生成新的 UUID，因此直接重新运行会创建一个新的逻辑请求。
- 2.
恢复现有任务
如果已有 task_id，请获取该任务。如果创建请求在收到 ID 之前超时，请使用已保存且未经更改的任务请求体和 requestId 再次创建。API 会返回现有任务；复用 ID 时更改输入会导致冲突。
- 3.
检查是否已经提交
该期限也涵盖向目标发送的 POST 请求。如果该请求超时，请在再次提交之前检查应用处理结果：即使客户端未收到响应，应用也可能已经收到请求。恢复 solvethree 任务并不能确定目标端发生了什么。
运行您的第一个任务。
使用 $5 工作区额度开始体验。