从 FastAPI 后端调用 solvethree

使用 Python 3.10 或更高版本。运行 python -m pip install "fastapi[standard]" 安装 FastAPI，然后运行 python -m pip install https://solvethree.com/downloads/solvethree_sdk-0.1.0-py3-none-any.whl 安装官方 SDK。

启动应用前，在服务器进程中设置以下环境变量：

• SOLVETHREE_API_KEY：工作区 API 密钥。请将其保留在服务器上；SDK 使用它向 solvethree 进行身份验证。
• INTEGRATION_KEY：独立的密钥，调用方通过 X-Integration-Key 请求头发送它，以向你的 FastAPI 接口进行身份验证。
• CAPTCHA_SITE：目标主机名，而不是完整 URL。
• SOLVETHREE_PROXY：此任务使用的 HTTP 或 HTTPS 代理 URL，不含路径、查询参数或片段。

调用接口前，先在工作区中为目标创建一个已启用的配置。示例将操作固定为 login，因此配置和预期的目标请求必须与该用途匹配。提供主机名不会创建配置；请参阅站点配置设置。

将此示例保存为 main.py。两个接口都要求提供 X-Integration-Key。创建接口完全根据服务器配置构建任务；调用方不在请求正文中提供站点、操作或代理。

# pip install "fastapi[standard]" and the official solvethree SDK wheel.
import hmac
import os
from fastapi import FastAPI, HTTPException, Request
from fastapi.concurrency import run_in_threadpool
from solvethree import Solvethree, SolvethreeError

app = FastAPI()

def authorize(request):
    if not hmac.compare_digest(request.headers.get("x-integration-key", ""), os.environ["INTEGRATION_KEY"]):
        raise HTTPException(status_code=401, detail="Unauthorized")

@app.post("/captcha/tasks")
async def create(request: Request):
    authorize(request)
    task = {"site": os.environ["CAPTCHA_SITE"], "action": "login", "proxy": os.environ["SOLVETHREE_PROXY"]}
    if request.headers.get("x-request-id"):
        task["requestId"] = request.headers["x-request-id"]
    def submit():
        return Solvethree(os.environ["SOLVETHREE_API_KEY"]).create_task(task, wait_ms=0, timeout=20)
    try:
        return await run_in_threadpool(submit)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid task input")
    except SolvethreeError as error:
        raise HTTPException(status_code=502, detail={"requestId": error.request_id, "taskId": error.task_id})

@app.get("/captcha/tasks/{task_id}")
async def retrieve(task_id: str, request: Request):
    authorize(request)
    def fetch():
        return Solvethree(os.environ["SOLVETHREE_API_KEY"]).get_task(task_id, wait_ms=15000, timeout=25)
    try:
        return await run_in_threadpool(fetch)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid task ID")
    except SolvethreeError as error:
        raise HTTPException(status_code=502, detail={"requestId": error.request_id, "taskId": error.task_id})

任务默认使用 reCAPTCHA Enterprise v3。它发送 site、action 和一个 proxy，不包含 type 或 url 字段。

Python SDK 是同步的。在 async def 中直接调用它会阻塞事件循环，因此两个处理函数都使用 run_in_threadpool。创建时使用 wait_ms=0，SDK 请求超时为 20 秒；查询时允许等待 15 秒，SDK 请求超时为 25 秒。两个处理函数都不会使用 client.solve 等待整个求解过程。

本地开发时，在文件所在目录运行 fastapi dev main.py。接口为 POST http://127.0.0.1:8000/captcha/tasks 和 GET http://127.0.0.1:8000/captcha/tasks/{task_id}。FastAPI 会发现 app 对象；无需另外注册 URLconf。

1. 1.

   使用稳定的请求 ID 创建一次任务

   发送 POST /captcha/tasks，并携带 X-Integration-Key 和为此逻辑任务保留的 X-Request-ID。请求 ID 必须包含 1–64 个 ASCII 字母、数字、下划线或连字符。接口将其作为 requestId 传给 solvethree。保存返回的任务 id。

2. 2.

   读取任务状态前先检查 HTTP 状态

   封装层在身份验证失败时返回 401，在出现 ValueError 输入错误时返回 400，在捕获 SolvethreeError 异常时返回 502。其 502 的 detail 包含 SDK 错误中的 requestId 和 taskId。请先处理该 HTTP 错误，再将响应视为任务。

3. 3.

   查询现有任务

   携带 X-Integration-Key 调用 GET /captcha/tasks/{task_id}。每次 HTTP 响应成功后检查 status。即使等待结束，查询仍可能返回尚未进入终态的任务；请再次查询同一任务，而不是创建新任务。

4. 4.

   恢复结果不确定的创建请求

   如果创建超时且你已有任务 ID，请查询该任务。否则，使用原始 X-Request-ID 和未更改的服务端任务输入重复发送 POST。对相同任务复用该 ID 会返回现有任务；更改输入会产生冲突。请单独保留原始 ID，不要用错误响应中的 requestId 替换它。

这些接口返回任务数据，不会向目标网站提交任何内容。status 为 ready 后，发起目标请求的组件必须使用 solution.token，而不是任务 ID。

通过创建该任务时使用的同一代理提交，并使用返回的 solution.userAgent 和 solution.secChUa 浏览器身份。请保留原始任务配置，不要假定当前环境中仍包含同一代理。SDK 的 successful_proxy(original_task, ready_result) 辅助函数会从该原始任务中选择代理。

确保提交始终对应预期的目标和操作。任务状态为 ready 表示 solvethree 已生成解决方案；只有目标的响应才能表明应用请求是否被接受。