SMSCode API开发指南：快速集成虚拟号码到你的应用

SMSCode API开发指南：快速集成虚拟号码到你的应用
SMSCode Team
SMSCode Team
·26 分钟阅读

如果你需要在应用中自动化处理SMS验证码接收，手动操作控制台明显不够高效。SMSCode提供完整的REST API，让你可以将虚拟号码能力直接集成到自己的系统中。

本文从零开始，带你完成SMSCode API的集成，包含可以直接使用的代码示例。

TL;DR： SMSCode公开API使用Bearer认证、REST endpoint和JSON响应。核心流程是：选择目录产品 → 创建订单 → 轮询或接收Webhook → 按订单能力取消。本文提供Python和Node.js的完整代码示例。

准备工作

在开始编写代码前，需要完成以下准备：

  1. 注册SMSCode账号并完成邮箱验证
  2. 充值账户余额（API调用需要余额支撑）
  3. 在控制台的“设置”→“API”页面获取你的API密钥
  4. 确认目标服务在SMSCode上可用（查看服务目录）

API密钥格式：64位十六进制字符串。请直接复制控制台生成的完整密钥，不要截断。

安全提示：API密钥等同于账号密码权限。不要将其写入代码仓库，应该通过环境变量注入。

API基础

Base URL

https://api.smscode.gg/v1

认证方式

所有请求需要在Header中携带API密钥：

Authorization: Bearer YOUR_API_KEY

不支持通过 URL 查询参数认证；请始终使用上面的 Authorization 请求头：

仅支持 Authorization: Bearer YOUR_API_KEY

请求/响应格式

所有API请求和响应均使用JSON格式。完整的API文档请访问开发者文档。

核心工作流

SMS验证码接收的完整流程如下：

1. 创建订单（解析订单ID；phone_number可能尚未分配）
    ↓
2. 有非空号码则直接使用；否则只做一次同ID的有界GET
    ↓
3. 号码已分配后才注册目标并轮询/等待验证码
    ↓
4. 按最新订单能力完成或取消

create已经解析为订单后必须保持resolved。create响应中的phone_number是optional/nullable；有效的 非空字符串直接使用且不发送GET，否则只允许调用现有有界GET一次。该GET后号码仍缺失、为null或 空白，或者发生identity不匹配、传输/解码失败时，立即返回带已解析order_id和status的 pending_assignment，并在打印号码、注册目标、轮询或取消之前停止。这个结果不是 needs_reconciliation，不得重放付费create，也不得附加或推断退款字段。

SMS consumer为每个订单把last_seen_revision初始化为-1。只有显式integer类型且严格大于调用方 状态的sms_revision，同时配有非空白字符串otp_message，才先消费消息并更新revision，然后再检查 终态；该规则不依赖OTP_RECEIVED或otp_code。boolean、旧/相同revision或空白消息都不得推进状态。

Python集成示例

以下是一个可以直接使用的Python集成示例：

下面的CreateAttemptStore不是普通键值存储。调用方必须先持久化非敏感的 caller_scope、唯一的business_job_id、Idempotency-Key和store时钟上的绝对人工复核时间点 review_threshold_at；prepare还要一次性持久化started_at、不可延长的 effective_replay_deadline_at = min(review_threshold_at, started_at + CLIENT_REPLAY_MAX)和 sends = 0。CLIENT_REPLAY_MAX = 3600与MAX_AUTOMATIC_SENDS = 32都是客户端付费重放的 安全预算，不是服务端时序承诺。store以 (caller_scope, business_job_id)为UNIQUE键，并把该键绑定到不可变endpoint与规范化body的 fingerprint。所有load、claim和CAS append都使用同一对键。每次worker只执行一次到期发送； needs_reconciliation的下次执行时间由store时钟和有界退避决定，而不是进程内sleep。 claim_for_send只能原子领取prepared、已到期的needs_reconciliation或lease已过期的 sending；即使进程在POST后崩溃，恢复worker也只能重放已存储的同一endpoint、key和body。 每次claim之前以及POST之前都要用store时钟和持久化的sends检查这两个边界；任一边界到达时， 通过CAS保留operator_review且不发送POST。不得换key、换body或创建替代订单。

import requests
import hashlib
import json
import math
import time
from urllib3.util import Timeout
from myapp.attempts import durable_create_attempt_store

DEFINITIVE_CREATE_ERRORS = {
    "NO_OFFER_AVAILABLE",
    "VALIDATION_ERROR",
    "PROVIDER_ERROR",
    "IDEMPOTENCY_KEY_REUSED",
}
CREATE_CONNECT_TIMEOUT_SECONDS = 5
CREATE_TOTAL_TIMEOUT_SECONDS = 30
CREATE_LEASE_SECONDS = 45
RECONCILIATION_INITIAL_BACKOFF_SECONDS = 1
RECONCILIATION_BACKOFF_CAP_SECONDS = 30
CLIENT_REPLAY_MAX = 3600
MAX_AUTOMATIC_SENDS = 32
ORDER_STATUSES = {"ACTIVE", "OTP_RECEIVED", "COMPLETED", "CANCELED", "EXPIRED"}
CREATE_ITEM_REQUIRED_FIELDS = {
    "id", "status", "product_id", "amount", "otp_code", "can_finish",
    "can_resend", "can_cancel", "can_replace", "can_reactivate",
    "resend_available_at", "cancel_available_at", "replace_available_at",
}
CREATE_ITEM_OPTIONAL_FIELDS = {
    "phone_number", "otp_received_at", "expires_at", "failed_reason",
    "catalog_product_id", "operator_id", "operator_name",
}
CANCEL_ERROR_CODES_BY_STATUS = {
    401: {"UNAUTHORIZED"},
    404: {"NOT_FOUND"},
    409: {"CONFLICT", "CANCEL_TOO_EARLY"},
    422: {"PROVIDER_ERROR"},
    429: {"RATE_LIMIT_EXCEEDED"},
}


def transport_timeout() -> Timeout:
    return Timeout(
        connect=CREATE_CONNECT_TIMEOUT_SECONDS,
        total=CREATE_TOTAL_TIMEOUT_SECONDS,
    )


def is_int32(value) -> bool:
    return type(value) is int and -(2 ** 31) <= value <= 2 ** 31 - 1


def is_int64(value) -> bool:
    return type(value) is int and -(2 ** 63) <= value <= 2 ** 63 - 1


def is_nullable_string(value) -> bool:
    return value is None or isinstance(value, str)


def validate_v1_create_order_item(item) -> bool:
    if not isinstance(item, dict):
        return False
    keys = set(item)
    allowed = CREATE_ITEM_REQUIRED_FIELDS | CREATE_ITEM_OPTIONAL_FIELDS
    if not CREATE_ITEM_REQUIRED_FIELDS <= keys or not keys <= allowed:
        return False
    if not is_int32(item["id"]) or not is_int32(item["product_id"]):
        return False
    if (
        not isinstance(item["status"], str)
        or item["status"] not in ORDER_STATUSES
        or not is_int64(item["amount"])
    ):
        return False
    if not is_nullable_string(item["otp_code"]):
        return False
    for field in (
        "can_finish", "can_resend", "can_cancel", "can_replace", "can_reactivate"
    ):
        if type(item[field]) is not bool:
            return False
    for field in (
        "resend_available_at", "cancel_available_at", "replace_available_at",
        "phone_number", "otp_received_at", "expires_at", "failed_reason",
        "operator_name",
    ):
        if field in item and not is_nullable_string(item[field]):
            return False
    for field in ("catalog_product_id", "operator_id"):
        if field in item and item[field] is not None and not is_int32(item[field]):
            return False
    return True


def is_v1_error_response(value) -> bool:
    if (
        type(value) is not dict
        or value.get("success") is not False
        or set(value) != {"success", "error"}
        or type(value["error"]) is not dict
    ):
        return False
    error = value["error"]
    fields = set(error)
    return (
        {"code", "message"} <= fields <= {"code", "message", "details"}
        and type(error["code"]) is str
        and type(error["message"]) is str
        and ("details" not in error or type(error["details"]) is dict)
    )


def validate_v1_create_order_result(data, quantity: int) -> bool:
    if not isinstance(data, dict) or set(data) != {"orders", "failed_count"}:
        return False
    if not isinstance(data["orders"], list):
        return False
    if not is_int32(data["failed_count"]) or data["failed_count"] < 0:
        return False
    if len(data["orders"]) + data["failed_count"] != quantity:
        return False
    return all(validate_v1_create_order_item(order) for order in data["orders"])


class SMSCodeClient:
    def __init__(self, api_key: str, attempt_store):
        self.api_key = api_key
        self.base_url = "https://api.smscode.gg/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }
        self.attempt_store = attempt_store

    def prepare_create_attempt(
        self,
        caller_scope: str,
        business_job_id: str,
        idempotency_key: str,
        review_threshold_at: int,
        catalog_product_id: int,
    ) -> dict:
        if not all(
            isinstance(value, str) and value.strip()
            for value in (caller_scope, business_job_id, idempotency_key)
        ):
            raise ValueError("caller scope、business job ID和幂等键必须由调用方持久化")
        if type(review_threshold_at) is not int or review_threshold_at < 0:
            raise ValueError("review_threshold_at必须是调用方配置的store时钟绝对时间点")
        if not is_int32(catalog_product_id):
            raise ValueError("catalog_product_id必须是int32")
        body = {"catalog_product_id": catalog_product_id, "quantity": 1}
        endpoint = f"{self.base_url}/orders/create"
        body_json = json.dumps(body, separators=(",", ":"), sort_keys=True)
        request_fingerprint = hashlib.sha256(
            f"{endpoint}\0{body_json}".encode("utf-8")
        ).hexdigest()
        started_at = self.attempt_store.now()
        if type(started_at) is not int or started_at < 0:
            raise ValueError("attempt store必须返回有效的非负整数时钟")
        candidate = {
            "caller_scope": caller_scope,
            "business_job_id": business_job_id,
            "endpoint": endpoint,
            "body_json": body_json,
            "request_fingerprint": request_fingerprint,
            "idempotency_key": idempotency_key,
            "review_threshold_at": review_threshold_at,
            "started_at": started_at,
            "effective_replay_deadline_at": min(
                review_threshold_at,
                started_at + CLIENT_REPLAY_MAX,
            ),
            "sends": 0,
            "state": "prepared",
            "attempt_history": [],
        }
        attempt = self.attempt_store.insert_or_load(
            caller_scope=caller_scope,
            business_job_id=business_job_id,
            request_fingerprint=request_fingerprint,
            candidate=candidate,
        )
        self._assert_attempt_binding(
            caller_scope,
            business_job_id,
            attempt,
            idempotency_key=idempotency_key,
            review_threshold_at=review_threshold_at,
            request_fingerprint=request_fingerprint,
        )
        return attempt

    def _assert_attempt_binding(
        self,
        caller_scope: str,
        business_job_id: str,
        attempt: dict,
        *,
        idempotency_key: str | None = None,
        review_threshold_at: int | None = None,
        request_fingerprint: str | None = None,
    ) -> None:
        if not isinstance(attempt, dict):
            raise ValueError("attempt必须是持久化object")
        if not all(
            isinstance(value, str) and value.strip()
            for value in (caller_scope, business_job_id)
        ):
            raise ValueError("caller scope和business job ID必须由调用方持久化")
        if idempotency_key is not None and not (
            isinstance(idempotency_key, str) and idempotency_key.strip()
        ):
            raise ValueError("Idempotency-Key必须是调用方持久化的非空字符串")
        if review_threshold_at is not None and (
            type(review_threshold_at) is not int or review_threshold_at < 0
        ):
            raise ValueError("review_threshold_at必须是有效的store时钟绝对时间点")
        if (
            attempt.get("caller_scope") != caller_scope
            or attempt.get("business_job_id") != business_job_id
        ):
            raise ValueError("attempt不属于当前caller/job")
        started_at = attempt.get("started_at")
        stored_review_threshold_at = attempt.get("review_threshold_at")
        if (
            type(started_at) is not int
            or started_at < 0
            or type(stored_review_threshold_at) is not int
            or stored_review_threshold_at < 0
        ):
            raise ValueError("attempt缺少store时钟的started_at或review threshold")
        if attempt.get("effective_replay_deadline_at") != min(
            stored_review_threshold_at,
            started_at + CLIENT_REPLAY_MAX,
        ):
            raise ValueError("attempt effective replay deadline不匹配")
        if type(attempt.get("sends")) is not int or attempt["sends"] < 0:
            raise ValueError("attempt sends必须是非负整数")
        endpoint = attempt.get("endpoint")
        body_json = attempt.get("body_json")
        if endpoint != f"{self.base_url}/orders/create" or not isinstance(
            body_json, str
        ):
            raise ValueError("attempt endpoint/body格式无效")
        try:
            persisted_body = json.loads(body_json)
        except (TypeError, ValueError) as error:
            raise ValueError("attempt body不是有效的规范JSON") from error
        if (
            not isinstance(persisted_body, dict)
            or set(persisted_body) != {"catalog_product_id", "quantity"}
            or not is_int32(persisted_body["catalog_product_id"])
            or persisted_body["quantity"] != 1
            or type(persisted_body["quantity"]) is not int
            or json.dumps(persisted_body, separators=(",", ":"), sort_keys=True)
            != body_json
        ):
            raise ValueError("attempt body不符合规范化create contract")
        fingerprint = hashlib.sha256(
            f"{endpoint}\0{body_json}".encode("utf-8")
        ).hexdigest()
        if fingerprint != attempt.get("request_fingerprint"):
            raise ValueError("attempt endpoint/body fingerprint不匹配")
        if request_fingerprint is not None and fingerprint != request_fingerprint:
            raise ValueError("同一business job不能更换create body")
        if idempotency_key is not None and attempt.get("idempotency_key") != idempotency_key:
            raise ValueError("同一business job不能更换Idempotency-Key")
        if (
            review_threshold_at is not None
            and attempt.get("review_threshold_at") != review_threshold_at
        ):
            raise ValueError("同一business job不能更换review threshold")

    def _append_create_event(
        self,
        caller_scope: str,
        business_job_id: str,
        attempt: dict,
        claim_token: str,
        event: dict,
    ) -> dict:
        return self.attempt_store.append_event(
            caller_scope=caller_scope,
            business_job_id=business_job_id,
            claim_token=claim_token,
            expected_version=attempt["version"],
            event=event,
        )

    def _create_recovery_fields(self, attempt: dict) -> dict:
        return {
            "endpoint": attempt["endpoint"],
            "body_json": attempt["body_json"],
            "idempotency_key": attempt["idempotency_key"],
            "caller_scope": attempt["caller_scope"],
            "business_job_id": attempt["business_job_id"],
            "review_threshold_at": attempt["review_threshold_at"],
            "started_at": attempt["started_at"],
            "effective_replay_deadline_at": attempt[
                "effective_replay_deadline_at"
            ],
            "sends": attempt["sends"],
        }

    def _review_bound_reason(self, attempt: dict, store_now: int) -> str | None:
        if store_now >= attempt["effective_replay_deadline_at"]:
            return "effective_replay_deadline_reached"
        if attempt["sends"] >= MAX_AUTOMATIC_SENDS:
            return "automatic_send_limit_reached"
        return None

    def _operator_review_event(
        self,
        attempt: dict,
        store_now: int,
        reason: str,
    ) -> dict:
        return {
            **self._create_recovery_fields(attempt),
            "state": "operator_review",
            "reason": reason,
            "retained_at": store_now,
        }

    def _bounded_retry_seconds(self, response, send_number: int) -> int:
        raw = response.headers.get("Retry-After", "") if response is not None else ""
        retry_after = 0
        if isinstance(raw, str) and raw.isdigit():
            try:
                retry_after = int(raw)
            except ValueError:
                pass
        fallback = min(
            RECONCILIATION_INITIAL_BACKOFF_SECONDS
            * (2 ** max(send_number - 1, 0)),
            RECONCILIATION_BACKOFF_CAP_SECONDS,
        )
        return min(retry_after, RECONCILIATION_BACKOFF_CAP_SECONDS) if retry_after > 0 else fallback

    def _schedule_create_reconciliation(
        self,
        caller_scope: str,
        business_job_id: str,
        attempt: dict,
        claim_token: str,
        *,
        reason: str,
        send_number: int,
        response=None,
        details: dict | None = None,
    ) -> dict:
        store_now = self.attempt_store.now()
        bound_reason = self._review_bound_reason(attempt, store_now)
        state = "operator_review" if bound_reason is not None else "needs_reconciliation"
        event = {
            **self._create_recovery_fields(attempt),
            "state": state,
            "reason": bound_reason or reason,
            "send_number": send_number,
            **(details or {}),
        }
        if bound_reason is not None:
            event["last_reconciliation_reason"] = reason
        if state == "needs_reconciliation":
            event["next_attempt_at"] = min(
                store_now + self._bounded_retry_seconds(response, send_number),
                attempt["effective_replay_deadline_at"],
            )
        updated = self._append_create_event(
            caller_scope, business_job_id, attempt, claim_token, event
        )
        return {"kind": state, "attempt": updated}

    def reconcile_create_attempt(
        self,
        caller_scope: str,
        business_job_id: str,
        idempotency_key: str,
        review_threshold_at: int,
    ) -> dict:
        attempt = self.attempt_store.load(
            caller_scope=caller_scope,
            business_job_id=business_job_id,
        )
        self._assert_attempt_binding(
            caller_scope,
            business_job_id,
            attempt,
            idempotency_key=idempotency_key,
            review_threshold_at=review_threshold_at,
        )
        return self.get_number(
            caller_scope,
            business_job_id,
            idempotency_key,
            review_threshold_at,
            attempt,
        )

    def get_number(
        self,
        caller_scope: str,
        business_job_id: str,
        idempotency_key: str,
        review_threshold_at: int,
        attempt: dict,
    ) -> dict:
        """发送一次到期的持久化尝试；后续调用仍使用同一key、endpoint和body。"""
        self._assert_attempt_binding(
            caller_scope,
            business_job_id,
            attempt,
            idempotency_key=idempotency_key,
            review_threshold_at=review_threshold_at,
        )
        pre_claim_now = self.attempt_store.now()
        bound_reason = self._review_bound_reason(attempt, pre_claim_now)
        if bound_reason is not None:
            # store必须先按expected_version做CAS并核对不可变恢复字段，再原子追加保留事件；
            # CAS冲突时不得claim或POST。
            updated = self.attempt_store.retain_operator_review(
                caller_scope=caller_scope,
                business_job_id=business_job_id,
                expected_version=attempt["version"],
                event=self._operator_review_event(
                    attempt,
                    pre_claim_now,
                    bound_reason,
                ),
            )
            return {"kind": "operator_review", "attempt": updated}
        claimed = self.attempt_store.claim_for_send(
            caller_scope=caller_scope,
            business_job_id=business_job_id,
            expected_version=attempt["version"],
            lease_seconds=CREATE_LEASE_SECONDS,
        )
        attempt = claimed["attempt"]
        claim_token = claimed["claim_token"]
        self._assert_attempt_binding(
            caller_scope,
            business_job_id,
            attempt,
            idempotency_key=idempotency_key,
            review_threshold_at=review_threshold_at,
        )
        pre_post_now = self.attempt_store.now()
        bound_reason = self._review_bound_reason(attempt, pre_post_now)
        if bound_reason is not None:
            updated = self._append_create_event(
                caller_scope,
                business_job_id,
                attempt,
                claim_token,
                self._operator_review_event(attempt, pre_post_now, bound_reason),
            )
            return {"kind": "operator_review", "attempt": updated}

        send_number = attempt["sends"] + 1
        last_send_at = pre_post_now
        attempt = self._append_create_event(
            caller_scope,
            business_job_id,
            attempt,
            claim_token,
            {
                **self._create_recovery_fields(attempt),
                "number": send_number,
                "state": "sending",
                "last_send_at": last_send_at,
                "transport_deadline_at": last_send_at + CREATE_TOTAL_TIMEOUT_SECONDS,
                "sends": send_number,
            },
        )
        if attempt.get("sends") != send_number:
            raise RuntimeError("store未在sending CAS中持久化sends；禁止发送POST")
        try:
            response = requests.post(
                attempt["endpoint"],
                headers={**self.headers, "Idempotency-Key": attempt["idempotency_key"]},
                data=attempt["body_json"],
                timeout=transport_timeout(),
            )
        except requests.RequestException as error:
            return self._schedule_create_reconciliation(
                caller_scope,
                business_job_id,
                attempt,
                claim_token,
                reason="transport_ambiguous",
                send_number=send_number,
                details={"exception_type": type(error).__name__},
            )

        if response.status_code >= 500:
            return self._schedule_create_reconciliation(
                caller_scope,
                business_job_id,
                attempt,
                claim_token,
                reason="http_ambiguous",
                send_number=send_number,
                response=response,
                details={"http_status": response.status_code},
            )

        try:
            payload = json.loads(response.content.decode("utf-8", errors="strict"))
        except (UnicodeDecodeError, ValueError) as error:
            return self._schedule_create_reconciliation(
                caller_scope,
                business_job_id,
                attempt,
                claim_token,
                reason="malformed_response",
                send_number=send_number,
                response=response,
                details={
                    "http_status": response.status_code,
                    "exception_type": type(error).__name__,
                },
            )

        if isinstance(payload, dict) and payload.get("success") is False:
            error = payload.get("error")
            raw_code = error.get("code") if isinstance(error, dict) else None
            code = raw_code if isinstance(raw_code, str) else None
            if not is_v1_error_response(payload):
                return self._schedule_create_reconciliation(
                    caller_scope,
                    business_job_id,
                    attempt,
                    claim_token,
                    reason="malformed_response",
                    send_number=send_number,
                    response=response,
                    details={"http_status": response.status_code, "error_code": code},
                )
            if response.status_code == 422 and code in DEFINITIVE_CREATE_ERRORS:
                updated = self._append_create_event(
                    caller_scope,
                    business_job_id,
                    attempt,
                    claim_token,
                    {
                        "state": "definitive_rejection",
                        "http_status": response.status_code,
                        "error_code": code,
                    },
                )
                return {"kind": "rejected", "error": error, "attempt": updated}
            if response.status_code == 409 and code == "INSUFFICIENT_BALANCE":
                updated = self._append_create_event(
                    caller_scope,
                    business_job_id,
                    attempt,
                    claim_token,
                    {
                        "state": "insufficient_balance",
                        "http_status": response.status_code,
                        "error_code": code,
                    },
                )
                return {"kind": "insufficient_balance", "error": error, "attempt": updated}
            reason = (
                "request_in_progress"
                if response.status_code == 409 and code == "REQUEST_IN_PROGRESS"
                else "response_ambiguous"
            )
            return self._schedule_create_reconciliation(
                caller_scope,
                business_job_id,
                attempt,
                claim_token,
                reason=reason,
                send_number=send_number,
                response=response,
                details={"http_status": response.status_code, "error_code": code},
            )

        body = json.loads(attempt["body_json"])
        result = payload.get("data") if isinstance(payload, dict) else None
        if (
            response.status_code != 200
            or not isinstance(payload, dict)
            or payload.get("success") is not True
            or "error" in payload
            or not validate_v1_create_order_result(result, body["quantity"])
        ):
            return self._schedule_create_reconciliation(
                caller_scope,
                business_job_id,
                attempt,
                claim_token,
                reason="contradictory_success",
                send_number=send_number,
                response=response,
                details={"http_status": response.status_code},
            )

        updated = self._append_create_event(
            caller_scope,
            business_job_id,
            attempt,
            claim_token,
            {"state": "resolved", "result": result},
        )
        return {"kind": "resolved", "result": result, "attempt": updated}

    def _decode_json_response(self, response) -> dict:
        payload = json.loads(response.content.decode("utf-8", errors="strict"))
        if not isinstance(payload, dict):
            raise ValueError("JSON envelope必须是object")
        return payload

    def _get_current_order(self, order_id: int) -> dict:
        if not is_int32(order_id):
            raise ValueError("order_id必须是int32")
        response = requests.get(
            f"{self.base_url}/orders/{order_id}",
            headers=self.headers,
            timeout=transport_timeout(),
        )
        response.raise_for_status()
        payload = self._decode_json_response(response)
        data = payload.get("data")
        if payload.get("success") is not True or not isinstance(data, dict):
            raise ValueError("订单快照envelope无效")
        return data

    def wait_for_sms(self, order_id: int, timeout: int = 120) -> str | None:
        """轮询等待验证码，返回验证码字符串或None"""
        if (
            isinstance(timeout, bool)
            or not isinstance(timeout, (int, float))
            or not math.isfinite(timeout)
        ):
            raise ValueError("timeout必须是有限数字")
        # `timeout` 限定的是本循环何时停止"发起"新一次轮询，而不是绝对耗时保证：
        # 已在途的请求仍会走完自身的 transport 上限。
        # 第一次轮询始终执行，即使预算为零或负数；`timeout` 限定的是何时停止
        # "发起"后续轮询，而不是绝对耗时保证。
        deadline = time.time() + timeout
        last_seen_revision = -1
        while True:
            data = self._get_current_order(order_id)
            sms_revision = data.get("sms_revision")
            otp_message = data.get("otp_message")

            if (
                type(sms_revision) is int
                and sms_revision > last_seen_revision
                and isinstance(otp_message, str)
                and otp_message.strip()
            ):
                last_seen_revision = sms_revision
                print(f"短信 revision {sms_revision}")
                return otp_message
            if data["status"] in ["COMPLETED", "CANCELED", "EXPIRED"]:
                return None

            # 先按剩余预算限时等待，再判断：睡眠本身可能耗尽预算，
            # 判断放在等待之后才能保证不会多发起一次轮询。
            time.sleep(min(5, max(0.0, deadline - time.time())))  # 本地配置，非API保证
            if time.time() >= deadline:
                break

        return None

    def _cancel_snapshot(self, order_id: int, data: dict) -> dict:
        if not isinstance(data, dict):
            raise ValueError("订单快照必须是object")
        status = data.get("status")
        if not isinstance(status, str) or status not in ORDER_STATUSES:
            raise ValueError("订单快照状态无效")
        if "id" not in data or not is_int32(data["id"]) or data["id"] != order_id:
            raise ValueError("订单快照ID无效")
        snapshot = {"id": data["id"], "status": status}
        if "can_cancel" in data and type(data["can_cancel"]) is bool:
            snapshot["can_cancel"] = data["can_cancel"]
        return snapshot

    def _valid_cancel_receipt(self, data, order_id: int) -> bool:
        return (
            isinstance(data, dict)
            and set(data) == {"order_id", "status", "refund_amount", "new_balance"}
            and data["order_id"] == order_id
            and is_int32(data["order_id"])
            and data["status"] == "CANCELED"
            and is_int64(data["refund_amount"])
            and data["refund_amount"] >= 0
            and is_int64(data["new_balance"])
        )

    def _reconcile_cancel(self, order_id: int, reason: str) -> dict:
        try:
            latest = self._get_current_order(order_id)
            snapshot = self._cancel_snapshot(order_id, latest)
        except (requests.RequestException, UnicodeDecodeError, ValueError, KeyError):
            return {"kind": "ambiguous", "order_id": order_id, "reason": reason}
        if snapshot["status"] == "CANCELED":
            return {"kind": "confirmed_canceled", "snapshot": snapshot}
        return {
            "kind": "ambiguous",
            "order_id": order_id,
            "reason": reason,
            "snapshot": snapshot,
        }

    def cancel_order(self, order_id: int) -> dict:
        """只返回skipped、receipt、rejected、confirmed_canceled或ambiguous。"""
        try:
            current = self._get_current_order(order_id)
            current_snapshot = self._cancel_snapshot(order_id, current)
        except (requests.RequestException, UnicodeDecodeError, ValueError, KeyError):
            return {"kind": "ambiguous", "order_id": order_id, "reason": "current_unavailable"}
        if type(current.get("can_cancel")) is not bool:
            return {
                "kind": "ambiguous",
                "order_id": order_id,
                "reason": "invalid_current_snapshot",
                "snapshot": current_snapshot,
            }
        if current["can_cancel"] and current_snapshot["status"] != "ACTIVE":
            return {
                "kind": "ambiguous",
                "order_id": order_id,
                "reason": "contradictory_current_snapshot",
                "snapshot": current_snapshot,
            }
        if not current["can_cancel"]:
            return {"kind": "skipped", "snapshot": current_snapshot}

        try:
            response = requests.post(
                f"{self.base_url}/orders/cancel",
                headers=self.headers,
                json={"id": order_id},
                timeout=transport_timeout(),
            )
        except requests.RequestException:
            return self._reconcile_cancel(order_id, "cancel_transport_ambiguous")
        try:
            payload = self._decode_json_response(response)
        except (UnicodeDecodeError, ValueError):
            return self._reconcile_cancel(order_id, "cancel_parse_ambiguous")

        data = payload.get("data")
        if payload.get("success") is True:
            if response.status_code == 200 and self._valid_cancel_receipt(data, order_id):
                return {"kind": "receipt", "receipt": data}
            return {
                "kind": "ambiguous",
                "order_id": order_id,
                "reason": "invalid_cancel_receipt",
            }
        error = payload.get("error")
        raw_code = error.get("code") if isinstance(error, dict) else None
        code = raw_code if isinstance(raw_code, str) else None
        if (
            payload.get("success") is False
            and code in CANCEL_ERROR_CODES_BY_STATUS.get(response.status_code, set())
        ):
            safe_error = {"code": code}
            if isinstance(error.get("message"), str):
                safe_error["message"] = error["message"]
            return {"kind": "rejected", "error": safe_error}
        return self._reconcile_cancel(order_id, "cancel_response_ambiguous")


# 使用示例：调用方先持久化四个job字段，再把它们原样传入workflow
def register_with_google_verification(
    caller_scope: str,
    business_job_id: str,
    idempotency_key: str,
    review_threshold_at: int,
    catalog_product_id: int,
):
    client = SMSCodeClient(
        api_key="YOUR_API_KEY",
        attempt_store=durable_create_attempt_store,
    )

    # 1. 使用目录中由调用方选择并持久化的产品坐标准备付费尝试。
    attempt = client.prepare_create_attempt(
        caller_scope=caller_scope,
        business_job_id=business_job_id,
        idempotency_key=idempotency_key,
        review_threshold_at=review_threshold_at,
        catalog_product_id=catalog_product_id,
    )
    create_outcome = client.get_number(
        caller_scope,
        business_job_id,
        idempotency_key,
        review_threshold_at,
        attempt,
    )
    if create_outcome["kind"] != "resolved":
        print(f"创建状态已持久化为 {create_outcome['kind']}；不要发起替代订单")
        return create_outcome
    if not create_outcome["result"]["orders"]:
        print("创建请求已解析，但没有创建订单")
        return create_outcome
    order = create_outcome["result"]["orders"][0]
    order_id = order["id"]
    phone_number = order.get("phone_number")
    assignment_status = order["status"]
    if not (isinstance(phone_number, str) and phone_number.strip()):
        try:
            current = client._get_current_order(order_id)
        except Exception:
            current = None
        if (
            isinstance(current, dict)
            and is_int32(current.get("id"))
            and current["id"] == order_id
            and isinstance(current.get("status"), str)
            and current["status"] in ORDER_STATUSES
            and "refund_amount" not in current
            and "new_balance" not in current
        ):
            assignment_status = current["status"]
            current_phone = current.get("phone_number")
            if isinstance(current_phone, str) and current_phone.strip():
                phone_number = current_phone

    if not (isinstance(phone_number, str) and phone_number.strip()):
        return {
            "create_outcome": "resolved",
            "assignment": {
                "kind": "pending_assignment",
                "order_id": order_id,
                "status": assignment_status,
            },
        }

    print(f"已获取号码: {phone_number}")
    print(f"订单ID: {order_id}")

    # 2. 将phone_number输入到Google注册表单...
    # （在这里添加你的Selenium/Playwright代码）

    # 3. 等待验证码
    code = client.wait_for_sms(order_id, timeout=120)

    if code:
        print(f"收到验证码: {code}")
        # 将验证码填入表单...
    else:
        cancellation = client.cancel_order(order_id)
        if cancellation["kind"] == "receipt":
            receipt = cancellation["receipt"]
            print(
                f"取消已确认；退款 Rp {receipt['refund_amount']}；"
                f"新余额 Rp {receipt['new_balance']}"
            )
        elif cancellation["kind"] == "confirmed_canceled":
            print("最新订单快照已确认CANCELED；该快照不是退款receipt")
        elif cancellation["kind"] == "skipped":
            print("can_cancel=false，未发送取消POST")
        elif cancellation["kind"] == "rejected":
            print(f"取消被API拒绝：{cancellation['error']['code']}")
        else:
            print("取消结果不明确；请持久化该outcome并继续核对")
        return cancellation

Node.js集成示例

const crypto = require('crypto');
const { Agent, request } = require('undici');
const { durableCreateAttemptStore } = require('./durable-create-attempt-store');

const DEFINITIVE_CREATE_ERRORS = new Set([
  'NO_OFFER_AVAILABLE',
  'VALIDATION_ERROR',
  'PROVIDER_ERROR',
  'IDEMPOTENCY_KEY_REUSED',
]);
const CREATE_CONNECT_TIMEOUT_MS = 5000;
const CREATE_TOTAL_TIMEOUT_MS = 30000;
const CREATE_LEASE_MS = 45000;
const RECONCILIATION_INITIAL_BACKOFF_MS = 1000;
const RECONCILIATION_BACKOFF_CAP_MS = 30000;
const CLIENT_REPLAY_MAX = 3600;
const MAX_AUTOMATIC_SENDS = 32;
const transportDispatcher = new Agent({ connectTimeout: CREATE_CONNECT_TIMEOUT_MS });
const ORDER_STATUSES = new Set([
  'ACTIVE', 'OTP_RECEIVED', 'COMPLETED', 'CANCELED', 'EXPIRED',
]);
const CREATE_ITEM_REQUIRED_FIELDS = new Set([
  'id', 'status', 'product_id', 'amount', 'otp_code', 'can_finish',
  'can_resend', 'can_cancel', 'can_replace', 'can_reactivate',
  'resend_available_at', 'cancel_available_at', 'replace_available_at',
]);
const CREATE_ITEM_OPTIONAL_FIELDS = new Set([
  'phone_number', 'otp_received_at', 'expires_at', 'failed_reason',
  'catalog_product_id', 'operator_id', 'operator_name',
]);
const CANCEL_ERROR_CODES_BY_STATUS = new Map([
  [401, new Set(['UNAUTHORIZED'])],
  [404, new Set(['NOT_FOUND'])],
  [409, new Set(['CONFLICT', 'CANCEL_TOO_EARLY'])],
  [422, new Set(['PROVIDER_ERROR'])],
  [429, new Set(['RATE_LIMIT_EXCEEDED'])],
]);

function canonicalJson(value) {
  if (Array.isArray(value)) {
    return `[${value.map(canonicalJson).join(',')}]`;
  }
  if (value !== null && typeof value === 'object') {
    return `{${Object.keys(value).sort().map((key) =>
      `${JSON.stringify(key)}:${canonicalJson(value[key])}`
    ).join(',')}}`;
  }
  return JSON.stringify(value);
}

function isRecord(value) {
  return value !== null && typeof value === 'object' && !Array.isArray(value);
}

function isV1ErrorResponse(value) {
  if (
    !isRecord(value) ||
    value.success !== false ||
    Object.keys(value).sort().join(',') !== 'error,success' ||
    !isRecord(value.error)
  ) {
    return false;
  }
  const errorFields = Object.keys(value.error).sort().join(',');
  return (
    (errorFields === 'code,message' || errorFields === 'code,details,message') &&
    typeof value.error.code === 'string' &&
    typeof value.error.message === 'string' &&
    (!('details' in value.error) || isRecord(value.error.details))
  );
}

function isInt32(value) {
  return Number.isInteger(value) && value >= -(2 ** 31) && value <= 2 ** 31 - 1;
}

function isInt64(value) {
  return Number.isSafeInteger(value);
}

function isNullableString(value) {
  return value === null || typeof value === 'string';
}

function validateV1CreateOrderItem(item) {
  if (!isRecord(item)) return false;
  const keys = Object.keys(item);
  const allowed = new Set([...CREATE_ITEM_REQUIRED_FIELDS, ...CREATE_ITEM_OPTIONAL_FIELDS]);
  if (
    [...CREATE_ITEM_REQUIRED_FIELDS].some((key) => !Object.hasOwn(item, key)) ||
    keys.some((key) => !allowed.has(key))
  ) return false;
  if (!isInt32(item.id) || !isInt32(item.product_id)) return false;
  if (!ORDER_STATUSES.has(item.status) || !isInt64(item.amount)) return false;
  if (!isNullableString(item.otp_code)) return false;
  for (const field of [
    'can_finish', 'can_resend', 'can_cancel', 'can_replace', 'can_reactivate',
  ]) {
    if (typeof item[field] !== 'boolean') return false;
  }
  for (const field of [
    'resend_available_at', 'cancel_available_at', 'replace_available_at',
    'phone_number', 'otp_received_at', 'expires_at', 'failed_reason',
    'operator_name',
  ]) {
    if (Object.hasOwn(item, field) && !isNullableString(item[field])) return false;
  }
  for (const field of ['catalog_product_id', 'operator_id']) {
    if (Object.hasOwn(item, field) && item[field] !== null && !isInt32(item[field])) {
      return false;
    }
  }
  return true;
}

function validateV1CreateOrderResult(data, quantity) {
  if (!isRecord(data)) return false;
  const keys = Object.keys(data);
  if (keys.length !== 2 || !keys.includes('orders') || !keys.includes('failed_count')) {
    return false;
  }
  if (!Array.isArray(data.orders) || !isInt32(data.failed_count) || data.failed_count < 0) {
    return false;
  }
  if (data.orders.length + data.failed_count !== quantity) return false;
  return data.orders.every(validateV1CreateOrderItem);
}

class SMSCodeClient {
  constructor(apiKey, attemptStore) {
    this.apiKey = apiKey;
    this.baseUrl = 'https://api.smscode.gg/v1';
    this.headers = {
      Authorization: `Bearer ${apiKey}`,
      'Content-Type': 'application/json',
    };
    this.attemptStore = attemptStore;
  }

  async prepareCreateAttempt(
    callerScope,
    businessJobId,
    idempotencyKey,
    reviewThresholdAt,
    catalogProductId,
  ) {
    if (![callerScope, businessJobId, idempotencyKey]
      .every((value) => typeof value === 'string' && value.trim())) {
      throw new TypeError('caller scope、business job ID和幂等键必须由调用方持久化');
    }
    if (!Number.isSafeInteger(reviewThresholdAt) || reviewThresholdAt < 0) {
      throw new TypeError('reviewThresholdAt必须是调用方配置的store时钟绝对时间点');
    }
    if (!isInt32(catalogProductId)) {
      throw new TypeError('catalogProductId必须是int32');
    }
    const body = { catalog_product_id: catalogProductId, quantity: 1 };
    const endpoint = `${this.baseUrl}/orders/create`;
    const bodyJson = canonicalJson(body);
    const requestFingerprint = crypto.createHash('sha256')
      .update(endpoint).update('\0').update(bodyJson).digest('hex');
    const startedAt = this.attemptStore.now();
    if (!Number.isSafeInteger(startedAt) || startedAt < 0) {
      throw new TypeError('attempt store必须返回有效的非负整数时钟');
    }
    const candidate = {
      callerScope,
      businessJobId,
      endpoint,
      bodyJson,
      requestFingerprint,
      idempotencyKey,
      reviewThresholdAt,
      startedAt,
      effectiveReplayDeadlineAt: Math.min(
        reviewThresholdAt,
        startedAt + CLIENT_REPLAY_MAX * 1000,
      ),
      sends: 0,
      state: 'prepared',
      attemptHistory: [],
    };
    const attempt = await this.attemptStore.insertOrLoad({
      callerScope,
      businessJobId,
      requestFingerprint,
      candidate,
    });
    this.assertAttemptBinding(callerScope, businessJobId, attempt, {
      idempotencyKey,
      reviewThresholdAt,
      requestFingerprint,
    });
    return attempt;
  }

  assertAttemptBinding(callerScope, businessJobId, attempt, expected = {}) {
    if (!isRecord(attempt)) throw new TypeError('attempt必须是持久化object');
    if (![callerScope, businessJobId]
      .every((value) => typeof value === 'string' && value.trim())) {
      throw new TypeError('caller scope和business job ID必须由调用方持久化');
    }
    if (
      expected.idempotencyKey !== undefined &&
      !(typeof expected.idempotencyKey === 'string' && expected.idempotencyKey.trim())
    ) throw new TypeError('Idempotency-Key必须是调用方持久化的非空字符串');
    if (
      expected.reviewThresholdAt !== undefined &&
      (!Number.isSafeInteger(expected.reviewThresholdAt) || expected.reviewThresholdAt < 0)
    ) throw new TypeError('reviewThresholdAt必须是有效的store时钟绝对时间点');
    if (
      attempt.callerScope !== callerScope ||
      attempt.businessJobId !== businessJobId
    ) throw new Error('attempt不属于当前caller/job');
    if (
      !Number.isSafeInteger(attempt.startedAt) ||
      attempt.startedAt < 0 ||
      !Number.isSafeInteger(attempt.reviewThresholdAt) ||
      attempt.reviewThresholdAt < 0
    ) throw new Error('attempt缺少store时钟的startedAt或review threshold');
    if (
      !Number.isSafeInteger(attempt.effectiveReplayDeadlineAt) ||
      attempt.effectiveReplayDeadlineAt !== Math.min(
        attempt.reviewThresholdAt,
        attempt.startedAt + CLIENT_REPLAY_MAX * 1000,
      )
    ) throw new Error('attempt effective replay deadline不匹配');
    if (!Number.isSafeInteger(attempt.sends) || attempt.sends < 0) {
      throw new Error('attempt sends必须是非负整数');
    }
    if (
      attempt.endpoint !== `${this.baseUrl}/orders/create` ||
      typeof attempt.bodyJson !== 'string'
    ) throw new Error('attempt endpoint/body格式无效');
    let persistedBody;
    try {
      persistedBody = JSON.parse(attempt.bodyJson);
    } catch {
      throw new Error('attempt body不是有效的规范JSON');
    }
    if (
      !isRecord(persistedBody) ||
      Object.keys(persistedBody).sort().join(',') !== 'catalog_product_id,quantity' ||
      !isInt32(persistedBody.catalog_product_id) ||
      persistedBody.quantity !== 1 ||
      canonicalJson(persistedBody) !== attempt.bodyJson
    ) throw new Error('attempt body不符合规范化create contract');
    const fingerprint = crypto.createHash('sha256')
      .update(attempt.endpoint).update('\0').update(attempt.bodyJson).digest('hex');
    if (fingerprint !== attempt.requestFingerprint) {
      throw new Error('attempt endpoint/body fingerprint不匹配');
    }
    if (
      expected.requestFingerprint !== undefined &&
      fingerprint !== expected.requestFingerprint
    ) throw new Error('同一business job不能更换create body');
    if (
      expected.idempotencyKey !== undefined &&
      attempt.idempotencyKey !== expected.idempotencyKey
    ) throw new Error('同一business job不能更换Idempotency-Key');
    if (
      expected.reviewThresholdAt !== undefined &&
      attempt.reviewThresholdAt !== expected.reviewThresholdAt
    ) throw new Error('同一business job不能更换review threshold');
  }

  async appendCreateEvent(callerScope, businessJobId, attempt, claimToken, event) {
    return this.attemptStore.appendEvent({
      callerScope,
      businessJobId,
      claimToken,
      expectedVersion: attempt.version,
      event,
    });
  }

  createRecoveryFields(attempt) {
    return {
      endpoint: attempt.endpoint,
      bodyJson: attempt.bodyJson,
      idempotencyKey: attempt.idempotencyKey,
      callerScope: attempt.callerScope,
      businessJobId: attempt.businessJobId,
      reviewThresholdAt: attempt.reviewThresholdAt,
      startedAt: attempt.startedAt,
      effectiveReplayDeadlineAt: attempt.effectiveReplayDeadlineAt,
      sends: attempt.sends,
    };
  }

  reviewBoundReason(attempt, storeNow) {
    if (storeNow >= attempt.effectiveReplayDeadlineAt) {
      return 'effective_replay_deadline_reached';
    }
    if (attempt.sends >= MAX_AUTOMATIC_SENDS) {
      return 'automatic_send_limit_reached';
    }
    return null;
  }

  operatorReviewEvent(attempt, storeNow, reason) {
    return {
      ...this.createRecoveryFields(attempt),
      state: 'operator_review',
      reason,
      retainedAt: storeNow,
    };
  }

  boundedRetryMs(response, sendNumber) {
    const raw = response?.headers?.['retry-after'];
    const parsed = typeof raw === 'string' && /^\d+$/.test(raw) ? Number(raw) * 1000 : 0;
    const fallback = Math.min(
      RECONCILIATION_INITIAL_BACKOFF_MS * (2 ** Math.max(sendNumber - 1, 0)),
      RECONCILIATION_BACKOFF_CAP_MS,
    );
    return parsed > 0 ? Math.min(parsed, RECONCILIATION_BACKOFF_CAP_MS) : fallback;
  }

  async scheduleCreateReconciliation(
    callerScope,
    businessJobId,
    attempt,
    claimToken,
    { reason, sendNumber, response, details = {} },
  ) {
    const storeNow = this.attemptStore.now();
    const boundReason = this.reviewBoundReason(attempt, storeNow);
    const state = boundReason === null ? 'needs_reconciliation' : 'operator_review';
    const event = {
      ...this.createRecoveryFields(attempt),
      state,
      reason: boundReason ?? reason,
      sendNumber,
      ...details,
    };
    if (boundReason !== null) event.lastReconciliationReason = reason;
    if (state === 'needs_reconciliation') {
      event.nextAttemptAt = Math.min(
        storeNow + this.boundedRetryMs(response, sendNumber),
        attempt.effectiveReplayDeadlineAt,
      );
    }
    const updated = await this.appendCreateEvent(
      callerScope, businessJobId, attempt, claimToken, event,
    );
    return { kind: state, attempt: updated };
  }

  async reconcileCreateAttempt(
    callerScope,
    businessJobId,
    idempotencyKey,
    reviewThresholdAt,
  ) {
    const attempt = await this.attemptStore.load({ callerScope, businessJobId });
    this.assertAttemptBinding(callerScope, businessJobId, attempt, {
      idempotencyKey,
      reviewThresholdAt,
    });
    return this.getNumber(
      callerScope,
      businessJobId,
      idempotencyKey,
      reviewThresholdAt,
      attempt,
    );
  }

  async getNumber(
    callerScope,
    businessJobId,
    idempotencyKey,
    reviewThresholdAt,
    attempt,
  ) {
    this.assertAttemptBinding(callerScope, businessJobId, attempt, {
      idempotencyKey,
      reviewThresholdAt,
    });
    const preClaimNow = this.attemptStore.now();
    let boundReason = this.reviewBoundReason(attempt, preClaimNow);
    if (boundReason !== null) {
      // store必须先按expectedVersion做CAS并核对不可变恢复字段，再原子追加保留事件；
      // CAS冲突时不得claim或POST。
      const updated = await this.attemptStore.retainOperatorReview({
        callerScope,
        businessJobId,
        expectedVersion: attempt.version,
        event: this.operatorReviewEvent(attempt, preClaimNow, boundReason),
      });
      return { kind: 'operator_review', attempt: updated };
    }
    const claimed = await this.attemptStore.claimForSend({
      callerScope,
      businessJobId,
      expectedVersion: attempt.version,
      leaseMs: CREATE_LEASE_MS,
    });
    attempt = claimed.attempt;
    const claimToken = claimed.claimToken;
    this.assertAttemptBinding(callerScope, businessJobId, attempt, {
      idempotencyKey,
      reviewThresholdAt,
    });
    const prePostNow = this.attemptStore.now();
    boundReason = this.reviewBoundReason(attempt, prePostNow);
    if (boundReason !== null) {
      const updated = await this.appendCreateEvent(
        callerScope,
        businessJobId,
        attempt,
        claimToken,
        this.operatorReviewEvent(attempt, prePostNow, boundReason),
      );
      return { kind: 'operator_review', attempt: updated };
    }

    const sendNumber = attempt.sends + 1;
    const lastSendAt = prePostNow;
    attempt = await this.appendCreateEvent(
      callerScope,
      businessJobId,
      attempt,
      claimToken,
      {
        ...this.createRecoveryFields(attempt),
        number: sendNumber,
        state: 'sending',
        lastSendAt,
        transportDeadlineAt: lastSendAt + CREATE_TOTAL_TIMEOUT_MS,
        sends: sendNumber,
      },
    );
    if (attempt.sends !== sendNumber) {
      throw new Error('store未在sending CAS中持久化sends；禁止发送POST');
    }

    let response;
    try {
      response = await request(attempt.endpoint, {
        method: 'POST',
        headers: { ...this.headers, 'Idempotency-Key': attempt.idempotencyKey },
        body: attempt.bodyJson,
        dispatcher: transportDispatcher,
        signal: AbortSignal.timeout(CREATE_TOTAL_TIMEOUT_MS),
      });
    } catch (error) {
      return this.scheduleCreateReconciliation(
        callerScope,
        businessJobId,
        attempt,
        claimToken,
        {
          reason: 'transport_ambiguous',
          sendNumber,
          details: { exceptionName: error?.name ?? 'Error' },
        },
      );
    }

    if (response.statusCode >= 500) {
      return this.scheduleCreateReconciliation(
        callerScope,
        businessJobId,
        attempt,
        claimToken,
        {
          reason: 'http_ambiguous',
          sendNumber,
          response,
          details: { httpStatus: response.statusCode },
        },
      );
    }

    let payload;
    try {
      payload = await this.readJsonResponse(response);
    } catch (error) {
      return this.scheduleCreateReconciliation(
        callerScope,
        businessJobId,
        attempt,
        claimToken,
        {
          reason: 'malformed_response',
          sendNumber,
          response,
          details: {
            httpStatus: response.statusCode,
            exceptionName: error?.name ?? 'Error',
          },
        },
      );
    }

    if (isRecord(payload) && payload.success === false) {
      const error = isRecord(payload.error) ? payload.error : {};
      const code = typeof error.code === 'string' ? error.code : null;
      if (!isV1ErrorResponse(payload)) {
        return this.scheduleCreateReconciliation(
          callerScope,
          businessJobId,
          attempt,
          claimToken,
          {
            reason: 'malformed_response',
            sendNumber,
            response,
            details: { httpStatus: response.statusCode, errorCode: code },
          },
        );
      }
      if (response.statusCode === 422 && DEFINITIVE_CREATE_ERRORS.has(code)) {
        const updated = await this.appendCreateEvent(
          callerScope,
          businessJobId,
          attempt,
          claimToken,
          { state: 'definitive_rejection', httpStatus: response.statusCode, errorCode: code },
        );
        return { kind: 'rejected', error, attempt: updated };
      }
      if (response.statusCode === 409 && code === 'INSUFFICIENT_BALANCE') {
        const updated = await this.appendCreateEvent(
          callerScope,
          businessJobId,
          attempt,
          claimToken,
          { state: 'insufficient_balance', httpStatus: response.statusCode, errorCode: code },
        );
        return { kind: 'insufficient_balance', error, attempt: updated };
      }
      const reason = response.statusCode === 409 && code === 'REQUEST_IN_PROGRESS'
        ? 'request_in_progress'
        : 'response_ambiguous';
      return this.scheduleCreateReconciliation(
        callerScope,
        businessJobId,
        attempt,
        claimToken,
        {
          reason,
          sendNumber,
          response,
          details: { httpStatus: response.statusCode, errorCode: code },
        },
      );
    }

    const body = JSON.parse(attempt.bodyJson);
    const result = isRecord(payload) ? payload.data : null;
    if (
      response.statusCode !== 200 ||
      !isRecord(payload) ||
      payload.success !== true ||
      'error' in payload ||
      !validateV1CreateOrderResult(result, body.quantity)
    ) {
      return this.scheduleCreateReconciliation(
        callerScope,
        businessJobId,
        attempt,
        claimToken,
        {
          reason: 'contradictory_success',
          sendNumber,
          response,
          details: { httpStatus: response.statusCode },
        },
      );
    }

    const updated = await this.appendCreateEvent(
      callerScope,
      businessJobId,
      attempt,
      claimToken,
      { state: 'resolved', result },
    );
    return { kind: 'resolved', result, attempt: updated };
  }

  async readJsonResponse(response) {
    const bytes = new Uint8Array(await response.body.arrayBuffer());
    const text = new TextDecoder('utf-8', { fatal: true }).decode(bytes);
    const payload = JSON.parse(text);
    if (!isRecord(payload)) throw new TypeError('JSON envelope必须是object');
    return payload;
  }

  async getCurrentOrder(orderId) {
    if (!isInt32(orderId)) throw new TypeError('orderId必须是int32');
    const response = await request(`${this.baseUrl}/orders/${orderId}`, {
      method: 'GET',
      headers: this.headers,
      dispatcher: transportDispatcher,
      signal: AbortSignal.timeout(CREATE_TOTAL_TIMEOUT_MS),
    });
    if (response.statusCode < 200 || response.statusCode >= 300) {
      throw new Error(`订单快照HTTP ${response.statusCode}`);
    }
    const payload = await this.readJsonResponse(response);
    if (payload.success !== true || !isRecord(payload.data)) {
      throw new Error('订单快照envelope无效');
    }
    return payload.data;
  }

  async waitForSms(orderId, timeout = 120000) {
    if (!Number.isFinite(timeout)) {
      throw new TypeError('timeout必须是有限数字');
    }
    // `timeout` 限定的是本循环何时停止"发起"新一次轮询，而不是绝对耗时保证。
    // 第一次轮询始终执行；`timeout` 限定的是何时停止"发起"后续轮询。
    const deadline = Date.now() + timeout;
    let lastSeenRevision = -1;
    while (true) {
      const data = await this.getCurrentOrder(orderId);
      const smsRevision = data.sms_revision;
      const otpMessage = data.otp_message;

      if (Number.isInteger(smsRevision) &&
          smsRevision > lastSeenRevision &&
          typeof otpMessage === 'string' && otpMessage.trim() !== '') {
        lastSeenRevision = smsRevision;
        console.log(`短信 revision ${smsRevision}`);
        return otpMessage;
      }
      if (['COMPLETED', 'CANCELED', 'EXPIRED'].includes(data.status)) {
        return null;
      }

      // 先按剩余预算限时等待，再判断：判断放在等待之后才不会多发起一次轮询。
      await new Promise(r =>
        setTimeout(r, Math.min(5000, Math.max(0, deadline - Date.now())))
      );
      if (Date.now() >= deadline) break;
    }
    return null;
  }

  cancelSnapshot(orderId, data) {
    if (!isRecord(data) || !ORDER_STATUSES.has(data.status)) {
      throw new TypeError('订单快照状态无效');
    }
    if (!Object.hasOwn(data, 'id') || !isInt32(data.id) || data.id !== orderId) {
      throw new TypeError('订单快照ID无效');
    }
    const snapshot = { id: data.id, status: data.status };
    if (typeof data.can_cancel === 'boolean') snapshot.can_cancel = data.can_cancel;
    return snapshot;
  }

  validCancelReceipt(data, orderId) {
    if (!isRecord(data)) return false;
    const keys = Object.keys(data).sort();
    const expected = ['new_balance', 'order_id', 'refund_amount', 'status'];
    return JSON.stringify(keys) === JSON.stringify(expected) &&
      isInt32(data.order_id) && data.order_id === orderId &&
      data.status === 'CANCELED' &&
      isInt64(data.refund_amount) && data.refund_amount >= 0 &&
      isInt64(data.new_balance);
  }

  async reconcileCancel(orderId, reason) {
    try {
      const latest = await this.getCurrentOrder(orderId);
      const snapshot = this.cancelSnapshot(orderId, latest);
      if (snapshot.status === 'CANCELED') {
        return { kind: 'confirmed_canceled', snapshot };
      }
      return { kind: 'ambiguous', orderId, reason, snapshot };
    } catch {
      return { kind: 'ambiguous', orderId, reason };
    }
  }

  async cancelOrder(orderId) {
    let current;
    let currentSnapshot;
    try {
      current = await this.getCurrentOrder(orderId);
      currentSnapshot = this.cancelSnapshot(orderId, current);
    } catch {
      return { kind: 'ambiguous', orderId, reason: 'current_unavailable' };
    }
    if (typeof current.can_cancel !== 'boolean') {
      return {
        kind: 'ambiguous',
        orderId,
        reason: 'invalid_current_snapshot',
        snapshot: currentSnapshot,
      };
    }
    if (current.can_cancel && currentSnapshot.status !== 'ACTIVE') {
      return {
        kind: 'ambiguous',
        orderId,
        reason: 'contradictory_current_snapshot',
        snapshot: currentSnapshot,
      };
    }
    if (!current.can_cancel) return { kind: 'skipped', snapshot: currentSnapshot };

    let response;
    try {
      response = await request(`${this.baseUrl}/orders/cancel`, {
        method: 'POST',
        headers: this.headers,
        body: JSON.stringify({ id: orderId }),
        dispatcher: transportDispatcher,
        signal: AbortSignal.timeout(CREATE_TOTAL_TIMEOUT_MS),
      });
    } catch {
      return this.reconcileCancel(orderId, 'cancel_transport_ambiguous');
    }
    let payload;
    try {
      payload = await this.readJsonResponse(response);
    } catch {
      return this.reconcileCancel(orderId, 'cancel_parse_ambiguous');
    }
    if (payload.success === true) {
      if (
        response.statusCode === 200 &&
        this.validCancelReceipt(payload.data, orderId)
      ) {
        return { kind: 'receipt', receipt: payload.data };
      }
      return { kind: 'ambiguous', orderId, reason: 'invalid_cancel_receipt' };
    }
    const code = isRecord(payload.error) && typeof payload.error.code === 'string'
      ? payload.error.code
      : null;
    if (
      payload.success === false &&
      CANCEL_ERROR_CODES_BY_STATUS.get(response.statusCode)?.has(code)
    ) {
      const error = { code };
      if (typeof payload.error.message === 'string') error.message = payload.error.message;
      return { kind: 'rejected', error };
    }
    return this.reconcileCancel(orderId, 'cancel_response_ambiguous');
  }
}

// 使用示例：job来自调用方的持久化任务表，重启后仍传入相同字段。
async function main(job) {
  const client = new SMSCodeClient(
    process.env.SMSCODE_API_KEY,
    durableCreateAttemptStore
  );

  const attempt = await client.prepareCreateAttempt(
    job.callerScope,
    job.businessJobId,
    job.idempotencyKey,
    job.reviewThresholdAt,
    job.catalogProductId,
  );
  const createOutcome = await client.getNumber(
    job.callerScope,
    job.businessJobId,
    job.idempotencyKey,
    job.reviewThresholdAt,
    attempt,
  );
  if (createOutcome.kind !== 'resolved') {
    console.log(`创建状态已持久化为 ${createOutcome.kind}；不要发起替代订单`);
    return createOutcome;
  }
  if (createOutcome.result.orders.length === 0) {
    console.log('创建请求已解析，但没有创建订单');
    return createOutcome;
  }
  const order = createOutcome.result.orders[0];
  let phoneNumber = order.phone_number;
  let assignmentStatus = order.status;
  if (!(typeof phoneNumber === 'string' && phoneNumber.trim() !== '')) {
    try {
      const current = await client.getCurrentOrder(order.id);
      if (isRecord(current) && isInt32(current.id) && current.id === order.id &&
          ORDER_STATUSES.has(current.status) &&
          !Object.hasOwn(current, 'refund_amount') &&
          !Object.hasOwn(current, 'new_balance')) {
        assignmentStatus = current.status;
        const currentPhone = current.phone_number;
        if (typeof currentPhone === 'string' && currentPhone.trim() !== '') {
          phoneNumber = currentPhone;
        }
      }
    } catch {
      // 查询失败时由下方返回pending_assignment。
    }
  }
  if (!(typeof phoneNumber === 'string' && phoneNumber.trim() !== '')) {
    return {
      createOutcome: 'resolved',
      assignment: {
        kind: 'pending_assignment',
        orderId: order.id,
        status: assignmentStatus,
      },
    };
  }

  console.log(`号码: ${phoneNumber}`);

  // 将号码输入WhatsApp注册页面...

  const code = await client.waitForSms(order.id);
  if (code) {
    console.log(`验证码: ${code}`);
  } else {
    const cancellation = await client.cancelOrder(order.id);
    if (cancellation?.kind === 'receipt') {
      console.log(
        `取消已确认；退款 Rp ${cancellation.receipt.refund_amount}；` +
        `新余额 Rp ${cancellation.receipt.new_balance}`
      );
    } else if (cancellation.kind === 'confirmed_canceled') {
      console.log('最新订单快照已确认CANCELED；该快照不是退款receipt');
    } else if (cancellation.kind === 'skipped') {
      console.log('can_cancel=false，未发送取消POST');
    } else if (cancellation.kind === 'rejected') {
      console.log(`取消被API拒绝：${cancellation.error.code}`);
    } else {
      console.log('取消结果不明确；请持久化该outcome并继续核对');
    }
    return cancellation;
  }
}

Webhook实时推送（推荐）

轮询方式简单，但每次请求都有网络开销。对于高并发场景，推荐使用Webhook——验证码到达时，SMSCode主动推送到你指定的URL。

配置Webhook

在 Account → Webhook Notifications 区域填写Webhook URL；保存后系统自动生成secret，并可用Send Test检查endpoint：

https://your-server.com/smscode-webhook

Webhook接收端（Node.js示例）

const express = require('express');
const crypto = require('crypto');
const { durableInbox } = require('./durable-inbox');
const app = express();

function verifyWebhookSignature(rawBody, signature) {
  const expected = `sha256=${crypto
    .createHmac('sha256', process.env.SMSCODE_WEBHOOK_SECRET)
    .update(rawBody)
    .digest('hex')}`;
  const actualBuffer = Buffer.from(signature);
  const expectedBuffer = Buffer.from(expected);
  return actualBuffer.length === expectedBuffer.length &&
    crypto.timingSafeEqual(actualBuffer, expectedBuffer);
}

function normalizeWebhookEvent(event, data) {
  if (event === 'webhook.test') {
    return { event, message: data.message };
  }
  return {
    event,
    orderId: data.order_id,
    message: data.otp_message,
    revision: data.sms_revision,
  };
}

function webhookDedupeKey(event, data, rawBody) {
  if (event === 'webhook.test') {
    return `${event}:${crypto.createHash('sha256').update(rawBody).digest('hex')}`;
  }
  if (event === 'order.otp_received') {
    return `${event}:${data.order_id}:${data.sms_revision}`;
  }
  return `${event}:${data.order_id}`;
}

async function persistOrEnqueueOnce(dedupeKey, eventRecord) {
  await durableInbox.insertIfAbsent(dedupeKey, eventRecord);
}

app.post(
  '/smscode-webhook',
  express.raw({ type: 'application/json' }),
  async (req, res) => {
  const signature = req.get('X-Webhook-Signature') || '';
  if (!verifyWebhookSignature(req.body, signature)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  const { event, data } = JSON.parse(req.body.toString('utf8'));
  const dedupeKey = webhookDedupeKey(event, data, req.body);
  const eventRecord = normalizeWebhookEvent(event, data);
  await persistOrEnqueueOnce(dedupeKey, eventRecord);
  return res.status(200).json({ received: true });
  }
);

insertIfAbsent 必须在唯一去重键上原子执行：OTP 事件使用 event + order_id + sms_revision，终态事件使用 event + order_id，webhook.test 使用原始请求体的 SHA-256。请求处理器只负责规范化事件并写入持久 inbox；业务副作用由独立 worker 从 inbox 执行。持久化失败时不要返回 2xx。

Webhook可以减少轮询请求，但仍可能重试或延迟。处理端必须使用原子去重键，并在持久化或 入队成功后才返回2xx；轮询仍用于对账。

错误处理最佳实践

常见错误码

错误码 HTTP 处理方式
UNAUTHORIZED 401 检查Bearer token
NOT_FOUND 404 检查资源ID和认证账户
INSUFFICIENT_BALANCE 409 一次POST后停止并充值
VALIDATION_ERROR 422 修正请求参数或catalog ID
RATE_LIMIT_EXCEEDED 429 正数Retry-After按客户端本地策略最多等待60秒；无效时备用延迟最多30秒，并服从外层deadline
SERVICE_UNAVAILABLE 503 付费create结果按不明确处理，先完成对账

重试逻辑

import time
import random
import requests
from urllib3.util import Timeout

POLL_CONNECT_TIMEOUT_SECONDS = 5
POLL_TOTAL_TIMEOUT_SECONDS = 30

def poll_with_retry(url, headers, max_retries=3, overall_timeout_seconds=120):
    deadline = time.monotonic() + overall_timeout_seconds
    for attempt in range(max_retries):
        remaining_seconds = deadline - time.monotonic()
        if remaining_seconds <= 0:
            raise TimeoutError("polling outer deadline exceeded")
        request_total_seconds = min(POLL_TOTAL_TIMEOUT_SECONDS, remaining_seconds)
        response = requests.get(
            url,
            headers=headers,
            timeout=Timeout(
                connect=min(POLL_CONNECT_TIMEOUT_SECONDS, request_total_seconds),
                total=request_total_seconds,
            ),
        )
        if response.status_code != 429:
            response.raise_for_status()
            return response
        if attempt == max_retries - 1:
            response.raise_for_status()
        raw_retry_after = response.headers.get("Retry-After", "")
        retry_after = int(raw_retry_after) if raw_retry_after.isdigit() else 0
        wait = (
            min(retry_after, 60)
            if retry_after > 0
            else min((2 ** attempt) + random.random(), 30)
        )
        remaining_seconds = deadline - time.monotonic()
        if remaining_seconds <= 0:
            raise TimeoutError("polling outer deadline exceeded")
        time.sleep(min(wait, remaining_seconds))
    raise RuntimeError("bounded polling retry exhausted")

这里的60秒是客户端本地保护上限，不是服务端SLA；缺失或无效header的备用延迟仍最多30秒。 overall_timeout_seconds形成绝对外层deadline，每次请求前都重新检查，单次请求timeout和sleep都按 剩余时间收紧，因此deadline到达后不会再发送请求。

这个通用退避助手只用于无付费副作用的GET轮询。付费create的恢复worker必须从任务表读取原来的 caller_scope、business_job_id、idempotency_key和绝对review_threshold_at，再调用上文的 reconcile_create_attempt；不得把create POST传给poll_with_retry，也不得生成替代key或body。

从目录选择产品

公开v1创建接口不接受service字符串。先调用 GET /v1/catalog/products?country_id=...&platform_id=...，从当前可用结果中选择产品； 创建请求必须且只能携带返回的product_id或非空catalog_product_id之一。目录价格和 可用量会变化，因此不要维护静态服务代码表，也不要硬编码产品坐标。

批量操作优化

对于需要同时处理大量号码的企业用户，需要考虑并发控制：

import asyncio

async def process_batch(
    client,
    jobs: list[tuple[str, str, str, int, dict]],
    concurrency: int = 10,
):
    semaphore = asyncio.Semaphore(concurrency)

    async def process_one(
        caller_scope: str,
        business_job_id: str,
        idempotency_key: str,
        review_threshold_at: int,
        prepared_attempt: dict,
    ):
        async with semaphore:
            create_outcome = await asyncio.to_thread(
                client.get_number,
                caller_scope,
                business_job_id,
                idempotency_key,
                review_threshold_at,
                prepared_attempt,
            )
            if create_outcome["kind"] != "resolved":
                return create_outcome
            if not create_outcome["result"]["orders"]:
                return create_outcome
            order = create_outcome["result"]["orders"][0]
            code = await asyncio.to_thread(client.wait_for_sms, order['id'])
            return {"order": order, "code": code}

    tasks = [
        process_one(
            caller_scope,
            business_job_id,
            idempotency_key,
            review_threshold_at,
            prepared_attempt,
        )
        for (
            caller_scope,
            business_job_id,
            idempotency_key,
            review_threshold_at,
            prepared_attempt,
        ) in jobs
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

每个jobs元素都来自持久化任务表，并携带caller_scope、business_job_id、持久化的 idempotency_key、绝对review_threshold_at和已经由prepare_create_attempt返回的attempt。 准备动作在入队前完成；worker不得只收到产品ID后临时生成业务任务ID、幂等键或复核阈值。

在应用中设置有界并发、轮询间隔和绝对总deadline。收到429时，正数Retry-After按客户端本地 策略最多等待60秒；如果缺失或无效，备用延迟最多30秒。两者都必须收紧到deadline剩余时间，且 deadline到达后不得再发请求；60秒不是服务端SLA，也不要假设固定账户配额。

迁移自SMS-Activate

SMSCode公开API并不是SMS-Activate协议的直接兼容替代，不能只修改base URL。现有集成至少需要调整以下部分：

  1. 将query string中的API key改为Authorization: Bearer ...。
  2. 将action调用改为SMSCode的/v1 REST endpoint。
  3. 将文本响应处理改为SMSCode的JSON envelope。
  4. 将服务、国家和activation ID映射为目录产品、订单ID和正式生命周期状态。

两者的业务流程都包括选择产品、创建订单和等待短信，但传输协议不同。请以上文的SMSCode原生REST示例为准，不要复用SMS-Activate的请求或状态代码。

查看SMSCode vs SMS-Activate的完整对比了解更多迁移细节。

还有问题？查看完整API文档，或者联系技术支持获取帮助。

FAQ

SMSCode API有免费试用额度吗？

SMSCode的创建订单请求会产生实际费用。测试前请查看账户充值页面和实时目录价格，并在每次 付费创建结果明确或完成对账后再发起下一次创建；不要假设存在免费额度、固定最低充值额或自动退款。

API的请求频率限制是多少？

请在应用中设置有界的并发、轮询间隔和绝对总deadline。收到 429 时，有效的正数 Retry-After按客户端本地策略最多等待60秒；该响应头缺失或无效时，备用延迟最多30秒。 所有等待都收紧到deadline剩余时间，deadline到达后不再发请求。60秒不是服务端SLA，也不要把 本地测试结果当作所有账户都适用的固定配额。

API密钥泄露了怎么办？

立即在控制台的API设置页面重置API密钥，旧密钥会立即失效。同时检查账号余额和订单历史，确认是否有异常消费。

SMSCode的API支持哪些编程语言？

SMSCode API是标准的REST API，任何支持HTTP请求的语言都可以使用：Python、Node.js、PHP、Java、Go、Ruby等。本文提供了Python和Node.js示例，其他语言的集成方式类似。

如何在测试环境中安全地使用API密钥？

使用环境变量存储API密钥（SMSCODE_API_KEY），不要硬编码到代码中。对于CI/CD环境，使用对应平台的Secrets功能（GitHub Secrets、GitLab CI Variables等）。测试环境建议使用单独的、余额有限的测试账号。

#api#开发者

准备试试 SMSCode？

创建账户，两分钟内获取第一个虚拟号码。

立即开始 →