Alibaba Wan Video

Alibaba Wan video

Use this skill to choose and operate the correct Wan surface. Treat hosted model IDs and downloadable checkpoints as different products even when their version numbers look related.

Route first

1. Ask for the task, inputs, output duration/resolution/aspect ratio, audio requirement, deployment region/data boundary, budget, latency, GPU capacity, and whether managed inference is acceptable.
2. Choose one route:
   • Hosted Model Studio for current Wan 2.7 quality, 2–15 second clips, generated/custom audio, 1080P, managed scaling, reference-to-video, continuation, or hosted editing.
   • Official open weights for offline processing, weight access, reproducible checkpoint pinning, adaptation, or self-hosting. The current official public family is Wan 2.2 (plus Wan 2.1 legacy), not hosted Wan 2.7.
3. Never silently substitute a hosted alias, an older API protocol, a community quantization, or a third-party gateway.
4. Before any billable request or large checkpoint download, show the exact model/revision, region, inputs, output settings, estimated charge or storage, and obtain explicit approval. A dry run must make no network mutation.

Facts and prices below were verified 2026-07-10. Recheck the linked first-party pages at execution time; availability, aliases, snapshots, limits, and prices are volatile.

Keep the product families separate

Wan 2.2 A14B is a two-expert MoE: roughly 27B total parameters with about 14B active per denoising step. TI2V-5B uses a high-compression VAE and officially supports 1280x704/704x1280 output at 24 fps. These facts do not make 5B and A14B interchangeable quality tiers.

Hosted Model Studio

Region and endpoint contract

Region, API key, model availability, service deployment scope, and endpoint must match. Keys are not cross-region credentials.

Workspace-dedicated domains isolate traffic to the workspace and are the documented production recommendation. Do not infer data residency merely from the access URL: the service deployment scope determines inference location, while the access region determines storage location. Record both.

Wan 2.7 request shapes

All current Wan 2.7 video modes use asynchronous HTTP:

• Create: POST {base}/api/v1/services/aigc/video-generation/video-synthesis
• Required headers: Authorization: Bearer ..., Content-Type: application/json, X-DashScope-Async: enable
• Poll: GET {base}/api/v1/tasks/{task_id}
• States: PENDING, RUNNING, SUCCEEDED, FAILED, CANCELED, UNKNOWN
• Poll about every 15 seconds with bounded backoff. Save the first task_id; do not POST again merely because generation is slow.
• Task queries and signed video_url values expire after 24 hours. Download promptly. Successful output is MP4/H.264.

Prefer HTTP when one worker must support every 2.7 mode. The current SDK matrix is uneven: DashScope Python/Java T2V wrappers support Wan 2.6 and earlier only (minimum documented versions 1.25.8/2.22.6), while the 2.7 I2V and R2V wrappers require Python 1.25.16 or Java 2.22.14. SDK "synchronous" calls merely wrap and block on the asynchronous service; they do not change job duration, billing, or result expiry. Recheck versions before pinning dependencies.

Core T2V body:

{
  "model": "wan2.7-t2v-2026-06-12",
  "input": {
    "prompt": "Generate a single-shot video. Wide shot of a red paper kite rising above a quiet salt flat at dawn; slow dolly forward, soft wind and natural ambience.",
    "negative_prompt": "text overlays, logos, warped kite, abrupt cuts"
  },
  "parameters": {
    "resolution": "720P",
    "ratio": "16:9",
    "duration": 5,
    "prompt_extend": false,
    "watermark": true,
    "seed": 104729
  }
}

Wan 2.7 T2V accepts Chinese or English prompts up to 5,000 characters and negative prompts up to 500; longer text is truncated, so validate locally. duration is an integer 2–15 (default 5); resolution is 720P or 1080P (default 1080P); ratios are 16:9, 9:16, 1:1, 4:3, or 3:4. Prompt extension defaults true, watermark defaults false, and a fixed seed improves—but does not guarantee—reproducibility. Control single versus multi-shot in natural-language prompt structure; the old shot_type has no effect on 2.7.

Change only input/model for other 2.7 modes:

// First and last frame I2V. Use only first_frame for first-frame generation.
{
  "model": "wan2.7-i2v-2026-04-25",
  "input": {"prompt": "The lantern drifts downstream as dusk deepens.", "media": [
    {"type": "first_frame", "url": "https://assets.example/first.webp"},
    {"type": "last_frame", "url": "https://assets.example/last.webp"}
  ]},
  "parameters": {"resolution": "720P", "duration": 6, "prompt_extend": false, "watermark": true}
}

// Continuation. first_clip may be combined with last_frame when supported by the exact model docs.
{
  "model": "wan2.7-i2v-2026-04-25",
  "input": {"prompt": "Continue the same camera move and weather.", "media": [
    {"type": "first_clip", "url": "https://assets.example/source.mp4"}
  ]},
  "parameters": {"resolution": "720P", "duration": 10, "prompt_extend": false, "watermark": true}
}

// Reference-to-video. Refer to assets by their one-based Image/Video order in the prompt.
{
  "model": "wan2.7-r2v-2026-06-12",
  "input": {"prompt": "Video 1 enters and gives Image 1 the compass.", "media": [
    {"type": "reference_video", "url": "https://assets.example/person.mp4", "reference_voice": "https://assets.example/voice.wav"},
    {"type": "reference_image", "url": "https://assets.example/robot.webp"}
  ]},
  "parameters": {"resolution": "720P", "ratio": "16:9", "duration": 8, "prompt_extend": false, "watermark": true}
}

For I2V, an input image may be JPEG/JPG/PNG without alpha, BMP, or WEBP, up to 20 MB, 240–8,000 pixels per side, and aspect ratio 1:8–8:1. A continuation first_clip must be MP4/MOV, 2–10 seconds, up to 100 MB, 240–4,096 pixels per side, and aspect ratio 1:8–8:1. In continuation, duration means the final total video length, not the number of new seconds; output ratio follows the clip and the full output duration is billed.

For R2V, duration is 2–10 seconds when a reference video is present and 2–15 without one. It accepts at most one first frame plus 1–5 combined reference images/videos; a character reference must contain only one character. Reference images have the same 20 MB limits above. Reference videos are MP4/MOV, 1–30 seconds, up to 100 MB, and 240–4,096 pixels per side; reference_voice is WAV/MP3, 1–10 seconds, up to 15 MB. A supplied first frame overrides ratio. For video editing, use wan2.7-videoedit, one {"type":"video"} plus up to four reference_image assets; its billing includes both input and output video duration. Always re-open the exact API page for volatile constraints before upload.

Billable, dry-run-first T2V client

This complete example pins the Singapore snapshot, prints a redacted free plan by default, binds the exact request/workspace/output policy plus fresh pricing evidence to approval, creates at most one task, persists a crash-resumable sanitized ledger, downloads through an exact-host and public-IP policy without forwarding the API key, and validates before atomic no-overwrite publication. It deliberately does not accept arbitrary base URLs or output URLs.

#!/usr/bin/env python3
import argparse, hashlib, http.client, ipaddress, json, os, random, socket, ssl
import pathlib, subprocess, sys, tempfile, threading, time, uuid
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from urllib.parse import urlsplit
import requests

MODEL = "wan2.7-t2v-2026-06-12"
PRICE_USD_PER_SECOND = {"720P": Decimal("0.10"), "1080P": Decimal("0.15")}
REGION_SUFFIX = {"sg": "ap-southeast-1.maas.aliyuncs.com"}
TERMINAL = {"SUCCEEDED", "FAILED", "CANCELED", "UNKNOWN"}
MAX_JSON = 1024 * 1024
MAX_VIDEO = 1024 * 1024 * 1024
MAX_TOOL = 1024 * 1024
EXPECTED = {
    ("720P", "16:9"): (1280, 720), ("720P", "9:16"): (720, 1280),
    ("720P", "1:1"): (960, 960), ("720P", "4:3"): (1104, 832),
    ("720P", "3:4"): (832, 1104), ("1080P", "16:9"): (1920, 1080),
    ("1080P", "9:16"): (1080, 1920), ("1080P", "1:1"): (1440, 1440),
    ("1080P", "4:3"): (1648, 1248), ("1080P", "3:4"): (1248, 1648)}

class APIError(RuntimeError):
    def __init__(self, message, status=None, retry_after=None, body_sha256=None):
        super().__init__(message); self.status = status; self.retry_after = retry_after; self.body_sha256 = body_sha256

def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()

def fsync_directory(path):
    if os.name == "nt": return
    fd = os.open(path, os.O_RDONLY)
    try: os.fsync(fd)
    finally: os.close(fd)

def atomic_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as f:
            json.dump(value, f, ensure_ascii=False, indent=2, sort_keys=True)
            f.write("\n"); f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path); fsync_directory(path.parent)
    finally:
        if os.path.exists(tmp): os.unlink(tmp)

def read_record(path):
    if path.stat().st_size > MAX_JSON: raise RuntimeError("attempt record exceeds cap")
    value = json.loads(path.read_bytes())
    if not isinstance(value, dict): raise RuntimeError("attempt record must be an object")
    return value

def run_bounded(argv, timeout):
    process = subprocess.Popen(argv, shell=False, stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err, total, lock, overflow = bytearray(), bytearray(), [0], threading.Lock(), threading.Event()
    def drain(stream, bucket):
        while True:
            chunk = stream.read(65536)
            if not chunk: break
            with lock:
                room = max(0, MAX_TOOL - total[0]); bucket.extend(chunk[:room]); total[0] += len(chunk)
                if total[0] > MAX_TOOL: overflow.set()
            if overflow.is_set():
                try: process.kill()
                except OSError: pass
    threads = [threading.Thread(target=drain, args=(process.stdout, out), daemon=True),
               threading.Thread(target=drain, args=(process.stderr, err), daemon=True)]
    for thread in threads: thread.start()
    timed_out = False
    try: code = process.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True; process.kill(); code = process.wait()
    for thread in threads: thread.join()
    return code, bytes(out), bytes(err), timed_out, overflow.is_set()

def api_json(session, method, url, payload=None, timeout=(10, 60)):
    with session.request(method, url, json=payload, timeout=timeout, allow_redirects=False, stream=True) as response:
        raw = bytearray()
        for chunk in response.iter_content(65536):
            raw.extend(chunk)
            if len(raw) > MAX_JSON: raise RuntimeError("API response exceeds cap")
        digest = hashlib.sha256(raw).hexdigest()
        if not 200 <= response.status_code < 300:
            retry = response.headers.get("Retry-After")
            try: retry = min(60.0, max(0.0, float(retry)))
            except (TypeError, ValueError): retry = None
            raise APIError(f"API HTTP {response.status_code}", response.status_code, retry, digest)
        try: value = json.loads(raw)
        except json.JSONDecodeError as exc: raise APIError("API returned invalid JSON", response.status_code, body_sha256=digest) from exc
        if not isinstance(value, dict): raise RuntimeError("API response is not an object")
        return value

def safe_get(call, deadline, attempts=5):
    for attempt in range(attempts):
        try: return call()
        except APIError as exc:
            if exc.status != 429 and (exc.status is None or exc.status < 500): raise
            if attempt + 1 == attempts or time.monotonic() >= deadline: raise
            delay = exc.retry_after or min(20, 2 ** attempt) * (1 + random.random() * 0.25)
        except requests.RequestException:
            if attempt + 1 == attempts or time.monotonic() >= deadline: raise
            delay = min(20, 2 ** attempt) * (1 + random.random() * 0.25)
        if time.monotonic() + delay >= deadline: raise TimeoutError("poll retry would cross deadline")
        time.sleep(delay)

class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host, address):
        super().__init__(host, 443, timeout=120, context=ssl.create_default_context()); self.address = address
    def connect(self):
        sock = socket.create_connection((self.address, 443), self.timeout)
        try: self.sock = self._context.wrap_socket(sock, server_hostname=self.host)
        except BaseException:
            sock.close(); raise

def download_video(url, stage_path, allowed_hosts):
    parsed = urlsplit(url); host = (parsed.hostname or "").lower()
    if (parsed.scheme != "https" or parsed.username or parsed.password or parsed.fragment
            or parsed.port not in (None, 443) or not parsed.path or host not in allowed_hosts):
        raise RuntimeError("result URL is not an exact approved HTTPS host")
    try: ipaddress.ip_address(host); raise RuntimeError("IP-literal result host rejected")
    except ValueError: pass
    addresses = sorted({item[4][0] for item in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM)})
    if not addresses or any(not ipaddress.ip_address(address).is_global for address in addresses):
        raise RuntimeError("result host did not resolve only to public addresses")
    connection = PinnedHTTPSConnection(host, addresses[0]); created = False
    try:
        target = parsed.path + (("?" + parsed.query) if parsed.query else "")
        connection.request("GET", target, headers={"Accept": "video/mp4,application/octet-stream"})
        response = connection.getresponse()
        if 300 <= response.status < 400: raise RuntimeError("result redirect rejected")
        if response.status != 200: raise RuntimeError(f"result HTTP {response.status}")
        mime = response.getheader("Content-Type", "").split(";", 1)[0].lower()
        if mime not in {"video/mp4", "application/octet-stream"}: raise RuntimeError("result MIME rejected")
        length = response.getheader("Content-Length")
        if length and (not length.isdigit() or int(length) > MAX_VIDEO): raise RuntimeError("declared result size rejected")
        fd = os.open(stage_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600); created = True; total = 0
        with os.fdopen(fd, "wb") as f:
            while True:
                chunk = response.read(1024 * 1024)
                if not chunk: break
                total += len(chunk)
                if total > MAX_VIDEO: raise RuntimeError("result crossed byte cap")
                f.write(chunk)
            f.flush(); os.fsync(f.fileno())
        return stage_path
    except BaseException:
        if created and stage_path.exists(): stage_path.unlink()
        raise
    finally: connection.close()

def inspect_mp4(path, output, resolution, ratio, duration):
    with path.open("rb") as f: head = f.read(12)
    if len(head) < 12 or head[4:8] != b"ftyp": raise RuntimeError("MP4 signature rejected")
    args = ["ffprobe", "-v", "error", "-show_entries",
            "format=duration,size:stream=codec_type,codec_name,width,height,r_frame_rate", "-of", "json", str(path)]
    code, out, _, timed_out, overflow = run_bounded(args, 30)
    if code or timed_out or overflow: raise RuntimeError("bounded ffprobe failed")
    probe = json.loads(out); streams = probe.get("streams", []); videos = [s for s in streams if s.get("codec_type") == "video"]
    if len(videos) != 1 or videos[0].get("codec_name") != "h264": raise RuntimeError("expected one H.264 stream")
    width, height = videos[0].get("width"), videos[0].get("height")
    if (width, height) != EXPECTED[(resolution, ratio)]: raise RuntimeError("dimensions do not match approval")
    try: actual = Decimal(str(probe.get("format", {}).get("duration", "")))
    except InvalidOperation as exc: raise RuntimeError("invalid media duration") from exc
    if not actual.is_finite() or abs(actual - Decimal(duration)) > Decimal("0.75"):
        raise RuntimeError("media duration outside tolerance")
    code, _, _, timed_out, overflow = run_bounded(["ffmpeg", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
    if code or timed_out or overflow: raise RuntimeError("bounded full decode failed")
    digest = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""): digest.update(chunk)
    return {"path": str(output), "bytes": path.stat().st_size, "sha256": digest.hexdigest(),
            "media": {"duration": str(actual), "width": width, "height": height,
                      "videoCodec": "h264", "audioStreams": sum(s.get("codec_type") == "audio" for s in streams)}}

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--prompt", required=True)
    p.add_argument("--negative-prompt", default="text overlays, logos, deformed objects")
    p.add_argument("--region", choices=sorted(REGION_SUFFIX), default="sg")
    p.add_argument("--resolution", choices=sorted(PRICE_USD_PER_SECOND), default="720P")
    p.add_argument("--ratio", choices=["16:9", "9:16", "1:1", "4:3", "3:4"], default="16:9")
    p.add_argument("--duration", type=int, default=5)
    p.add_argument("--seed", type=int, default=104729)
    p.add_argument("--workspace-id", required=True, help="non-secret Singapore Model Studio workspace ID")
    p.add_argument("--out-dir", type=pathlib.Path, default=pathlib.Path("wan-run"))
    p.add_argument("--attempt-id", required=True)
    p.add_argument("--max-usd", required=True)
    p.add_argument("--pricing-checked-at", required=True)
    p.add_argument("--pricing-evidence-sha256", required=True)
    p.add_argument("--delivery-host", action="append", required=True,
                   help="exact approved result host; repeat for each host")
    p.add_argument("--approval-sha256")
    p.add_argument("--execute", action="store_true"); p.add_argument("--resume", action="store_true")
    p.add_argument("--show-full-plan", action="store_true")
    a = p.parse_args()

    if not 1 <= len(a.prompt) <= 5000 or len(a.negative_prompt) > 500:
        p.error("prompt must be 1..5000 chars; negative prompt <=500")
    if not 2 <= a.duration <= 15 or not 0 <= a.seed <= 2147483647:
        p.error("duration must be 2..15 and seed 0..2147483647")
    try: attempt = uuid.UUID(a.attempt_id); ceiling = Decimal(a.max_usd)
    except (ValueError, InvalidOperation) as exc: p.error(f"attempt ID or maximum invalid: {exc}")
    if attempt.version != 4 or str(attempt) != a.attempt_id.lower(): p.error("attempt ID must be canonical UUIDv4")
    estimate = PRICE_USD_PER_SECOND[a.resolution] * a.duration
    if not ceiling.is_finite() or ceiling <= 0 or ceiling < estimate:
        p.error("maximum must be finite, positive, and cover the estimate")
    if (len(a.pricing_evidence_sha256) != 64
            or any(c not in "0123456789abcdef" for c in a.pricing_evidence_sha256)):
        p.error("pricing evidence must be a lowercase SHA-256")
    try: checked_at = datetime.fromisoformat(a.pricing_checked_at.replace("Z", "+00:00"))
    except ValueError as exc: p.error(f"pricing timestamp must be ISO-8601: {exc}")
    if checked_at.tzinfo is None: p.error("pricing timestamp must include a timezone")
    allowed_hosts = sorted(set(host.lower().rstrip(".") for host in a.delivery_host))
    if (not allowed_hosts or len(allowed_hosts) > 8
            or any(not host or len(host) > 253
                   or any(c not in "abcdefghijklmnopqrstuvwxyz0123456789-." for c in host) for host in allowed_hosts)):
        p.error("delivery hosts must be 1..8 exact DNS names")
    for host in allowed_hosts:
        try: ipaddress.ip_address(host); p.error("delivery hosts cannot be IP literals")
        except ValueError: pass
    out_dir = a.out_dir.resolve(); record_path = out_dir / "attempt.json"
    video_path = out_dir / "output.mp4"; stage_path = out_dir / ".output.mp4.wan-stage"
    body = {"model": MODEL, "input": {"prompt": a.prompt,
            "negative_prompt": a.negative_prompt}, "parameters": {
            "resolution": a.resolution, "ratio": a.ratio, "duration": a.duration,
            "prompt_extend": False, "watermark": True, "seed": a.seed}}
    workspace = a.workspace_id
    if (not workspace or len(workspace) > 100
            or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-" for c in workspace)):
        p.error("workspace ID contains unsupported characters")
    host = f"{workspace}.{REGION_SUFFIX[a.region]}"
    request_hash = hashlib.sha256(canonical(body)).hexdigest()
    envelope = {"backend": "alibaba-model-studio-hosted", "region": a.region,
        "serviceDeploymentScope": "International", "host": host,
        "endpoint": "/api/v1/services/aigc/video-generation/video-synthesis",
        "request": body, "attemptId": str(attempt), "attemptRecord": str(record_path),
        "outputPolicy": {"path": str(video_path), "stagePath": str(stage_path),
                         "exactDeliveryHosts": allowed_hosts, "maxBytes": MAX_VIDEO,
                         "mime": ["video/mp4", "application/octet-stream"], "noOverwrite": True,
                         "requiredCodec": "h264", "fullDecode": True},
        "cost": {"pricingCheckedAt": checked_at.isoformat(), "pricingEvidenceSha256": a.pricing_evidence_sha256,
                 "rateUsdPerOutputSecond": str(PRICE_USD_PER_SECOND[a.resolution]),
                 "seconds": a.duration, "creates": 1, "estimatedMaxUsd": str(estimate),
                 "approvedMaxUsd": str(ceiling)}}
    approval = hashlib.sha256(canonical(envelope)).hexdigest()
    safe_body = {**body, "input": {"prompt": "sha256:" + hashlib.sha256(a.prompt.encode()).hexdigest(),
                                    "negative_prompt": "sha256:" + hashlib.sha256(a.negative_prompt.encode()).hexdigest()}}
    print(json.dumps({"dryRun": not a.execute, "approvalSha256": approval,
        "plan": {**envelope, "request": safe_body}}, ensure_ascii=False, indent=2))
    if a.show_full_plan: print(json.dumps({"protectedFullPlan": envelope}, ensure_ascii=False, indent=2))
    if not a.execute:
        print("PLAN ONLY: no API request was made", file=sys.stderr)
        return 0
    if a.approval_sha256 != approval: raise SystemExit("exact approval digest mismatch")
    age = datetime.now(timezone.utc) - checked_at.astimezone(timezone.utc)
    if age.total_seconds() < -300 or age.total_seconds() > 86400:
        raise SystemExit("pricing evidence must be rechecked within 24 hours of execution")
    base = f"https://{workspace}.{REGION_SUFFIX[a.region]}"
    if a.resume:
        if not record_path.exists(): raise SystemExit("resume record missing")
        record = read_record(record_path)
        if record.get("approval_sha256") != approval or record.get("request_sha256") != request_hash:
            raise SystemExit("resume does not match exact approved request")
        task_id = record.get("task_id")
        if not isinstance(task_id, str) or not task_id: raise SystemExit("no known task ID; never replay create")
        expected_artifact = record.get("artifact_staged") or record.get("artifact")
        candidate = video_path if video_path.exists() else stage_path if stage_path.exists() else None
        if candidate is not None:
            if not isinstance(expected_artifact, dict):
                if record.get("status") == "downloading" and candidate == stage_path:
                    stage_path.unlink(); fsync_directory(stage_path.parent)
                else: raise SystemExit("unclaimed output/stage exists; refusing overwrite or adoption")
            else:
                artifact = inspect_mp4(candidate, video_path, a.resolution, a.ratio, a.duration)
                if (artifact.get("sha256") != expected_artifact.get("sha256")
                        or artifact.get("bytes") != expected_artifact.get("bytes")):
                    raise SystemExit("recovered artifact does not match durable staged evidence")
                if candidate == stage_path:
                    if video_path.exists(): raise SystemExit("output appeared during recovery")
                    os.replace(stage_path, video_path); fsync_directory(video_path.parent)
                record.update(status="artifact_saved", artifact=artifact, updated_unix=int(time.time()))
                record.pop("artifact_staged", None); atomic_json(record_path, record)
                print(json.dumps(record, indent=2)); return 0
        key = os.environ.get("DASHSCOPE_API_KEY")
        if not key: raise SystemExit("Set DASHSCOPE_API_KEY to resume polling")
        session = requests.Session(); session.headers.update({"Authorization": f"Bearer {key}",
            "Content-Type": "application/json", "X-DashScope-Async": "enable"})
    else:
        key = os.environ.get("DASHSCOPE_API_KEY")
        if not key: raise SystemExit("Set DASHSCOPE_API_KEY")
        session = requests.Session(); session.headers.update({"Authorization": f"Bearer {key}",
            "Content-Type": "application/json", "X-DashScope-Async": "enable"})
        out_dir.parent.mkdir(parents=True, exist_ok=True)
        try: out_dir.mkdir(exist_ok=False)
        except FileExistsError as exc: raise SystemExit("attempt directory already claimed; use resume") from exc
        record = {"attempt_id": str(attempt), "approval_sha256": approval, "request_sha256": request_hash,
                  "status": "posting", "task_id": None, "created_unix": int(time.time())}
        atomic_json(record_path, record)
        try: created = api_json(session, "POST", base + envelope["endpoint"], body, (10, 120))
        except Exception as exc:
            known_rejection = (isinstance(exc, APIError) and exc.status is not None and 400 <= exc.status < 500
                               and exc.status not in {408, 409, 425, 429})
            record.update(status="create_rejected" if known_rejection else "create_outcome_unknown",
                          error_type=type(exc).__name__, http_status=getattr(exc, "status", None),
                          error_body_sha256=getattr(exc, "body_sha256", None))
            atomic_json(record_path, record)
            if known_rejection: raise RuntimeError("provider rejected create before acceptance") from exc
            raise RuntimeError("create outcome unknown; reconcile billing/logs, never replay") from exc
        task_id = created.get("output", {}).get("task_id")
        if (not isinstance(task_id, str) or not 1 <= len(task_id) <= 200
                or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_." for c in task_id)):
            record.update(status="create_outcome_unknown", error_type="MissingOrInvalidTaskId",
                          create_response_sha256=hashlib.sha256(canonical(created)).hexdigest())
            atomic_json(record_path, record)
            raise RuntimeError("create returned no safe task ID; reconcile, never replay")
        record.update(status="task_id_saved", task_id=task_id, request_id=created.get("request_id"),
                      create_response_sha256=hashlib.sha256(canonical(created)).hexdigest(), updated_unix=int(time.time()))
        atomic_json(record_path, record)

    deadline = time.monotonic() + 12 * 60
    result = None
    while time.monotonic() < deadline:
        result = safe_get(lambda: api_json(session, "GET", base + f"/api/v1/tasks/{task_id}", timeout=(10, 30)), deadline)
        status = result.get("output", {}).get("task_status")
        print(f"task={task_id} status={status}", file=sys.stderr)
        if status in TERMINAL:
            break
        if status not in {"PENDING", "RUNNING"}: raise RuntimeError("unrecognized task status")
        if time.monotonic() + 15 >= deadline: break
        time.sleep(15)
    if result is None or result.get("output", {}).get("task_status") in {"PENDING", "RUNNING"}:
        record.update(status="polling_paused", updated_unix=int(time.time())); atomic_json(record_path, record)
        raise TimeoutError(f"Polling timed out; resume GET with task_id {task_id}, do not POST")
    output = result.get("output", {})
    if output.get("task_status") != "SUCCEEDED":
        message = str(output.get("message", ""))
        record.update(status="generation_" + str(output.get("task_status", "unknown")).lower(),
                      service_code=output.get("code"), message_sha256=hashlib.sha256(message.encode()).hexdigest())
        atomic_json(record_path, record); raise RuntimeError("task ended without a successful artifact")
    url = output.get("video_url", "")
    if not isinstance(url, str) or not url: raise RuntimeError("successful task omitted result URL")
    usage = result.get("usage") or {}; actual_cost = None
    try:
        seconds = Decimal(str(usage.get("output_video_duration")))
        if seconds.is_finite() and seconds >= 0: actual_cost = PRICE_USD_PER_SECOND[a.resolution] * seconds
    except InvalidOperation: pass
    record.update(status="downloading", result_response_sha256=hashlib.sha256(canonical(result)).hexdigest(),
                  result_url_sha256=hashlib.sha256(url.encode()).hexdigest(), usage=usage,
                  actual_cost_usd=None if actual_cost is None else str(actual_cost),
                  actual_cost_within_approval=None if actual_cost is None else actual_cost <= ceiling,
                  stage_path=str(stage_path), updated_unix=int(time.time()))
    atomic_json(record_path, record)
    try:
        download_video(url, stage_path, set(allowed_hosts))
        artifact = inspect_mp4(stage_path, video_path, a.resolution, a.ratio, a.duration)
    except BaseException as exc:
        record.update(status="artifact_validation_failed", error_type=type(exc).__name__)
        atomic_json(record_path, record); raise
    record.update(status="artifact_staged", artifact_staged=artifact, updated_unix=int(time.time()))
    atomic_json(record_path, record)
    if video_path.exists(): raise RuntimeError("output appeared before publication")
    os.replace(stage_path, video_path); fsync_directory(video_path.parent)
    record.update(status="artifact_saved", artifact=artifact, rights_review="required",
                  moderation_review="required", creative_qa="pending", ai_disclosure="required",
                  provenance_credentials="not_documented", updated_unix=int(time.time()))
    record.pop("artifact_staged", None); atomic_json(record_path, record)
    print(json.dumps(record, ensure_ascii=False, indent=2)); return 0

if __name__ == "__main__":
    main()

Install requests, then run the plan first. Only the second command can create a paid task:

python wan_hosted.py --workspace-id "$WAN_WORKSPACE_ID" --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.50 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$WAN_RESULT_HOST" --prompt "Generate a single-shot video. A red kite rises over a salt flat at dawn."
DASHSCOPE_API_KEY=... python wan_hosted.py --execute --approval-sha256 "<exact digest>" --workspace-id "$WAN_WORKSPACE_ID" --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.50 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$WAN_RESULT_HOST" --prompt "Generate a single-shot video. A red kite rises over a salt flat at dawn."

BILLING_SHA256 is the digest of the dated first-party pricing evidence actually reviewed. WAN_RESULT_HOST is an exact hostname allowed by the current Model Studio/OSS delivery contract, never a wildcard inferred from the returned URL. Repeat --delivery-host only for additional exact approved hosts. Execution rejects pricing evidence older than 24 hours. Use the unchanged arguments with --execute --resume only when the durable attempt record already contains a task ID; posting or create_outcome_unknown requires billing/log/support reconciliation and must never be replayed.

The executable deliberately accepts only --region sg. For Beijing or another region, first replace the region allowlist and price table with values from the current pricing page and re-review the endpoint/model contract; do not bypass that fail-closed boundary.

Cost and capacity controls

Current standard unit rates in USD per billable second:

Billing basis still differs by mode. I2V and T2V input is free, so their billable duration is the successful output duration. For Wan 2.7 R2V with N reference videos, each reference video is billed for min(actual duration, 5/N seconds) and those input amounts are added to the output duration; images and the first frame are excluded from N. Video editing bills input-video duration plus output-video duration. Failed requests are not billed and do not consume the free quota. Flash, legacy, and silent-output prices are different; consult the live table rather than extrapolating. Do not treat a promotional/free quota as a budget control.

Singapore and Beijing rate-limit the listed Wan 2.7 T2V/I2V/R2V/edit models at 5 task submissions/second and 5 concurrent tasks per Alibaba Cloud account. Limits aggregate RAM users, workspaces, and keys. A rate limit is not a cumulative spend limit: use account budget alerts, a local job queue, concurrency semaphore, and a per-job cost ceiling. Recheck because these values can change.

Official open-weight deployment

Use only official Wan-Video GitHub and Wan-AI Hugging Face/ModelScope artifacts unless the user explicitly accepts a community port or quantization. Record:

• repository URL and immutable Git commit;
• checkpoint repository and immutable revision/commit;
• every file hash, package lock, CUDA/PyTorch/driver version, GPU type, prompt, negative prompt, seed, size, frame count, sampling settings, and output hash;
• the license texts shipped with each pinned artifact and all dependency licenses.

Do not download weights as an incidental step. First run git ls-remote or inspect the registry metadata, present expected storage and license, get approval, then pin the reviewed hashes. The official Wan 2.2 repository has no GitHub release artifacts as of verification; main is not a production version.

Example after the operator has approved and pinned both revisions:

git clone https://github.com/Wan-Video/Wan2.2.git
cd Wan2.2
git checkout --detach "$REVIEWED_WAN_GIT_COMMIT"
python -m venv .venv
. .venv/bin/activate
python -m pip install --require-hashes -r requirements.lock
huggingface-cli download Wan-AI/Wan2.2-TI2V-5B \
  --revision "$REVIEWED_HF_COMMIT" --local-dir ./Wan2.2-TI2V-5B
sha256sum -c checkpoint.sha256
python generate.py --task ti2v-5B --size '1280*704' \
  --ckpt_dir ./Wan2.2-TI2V-5B --offload_model True --convert_model_dtype \
  --t5_cpu --base_seed 104729 --save_file ./wan-ti2v-5b.mp4 \
  --prompt "Wide shot of a red paper kite rising above a quiet salt flat at dawn."
ffprobe -v error -show_streams -show_format ./wan-ti2v-5b.mp4
ffmpeg -v error -i ./wan-ti2v-5b.mp4 -f null -

requirements.lock and checkpoint.sha256 are organization-generated review artifacts, not files supplied by the Wan repository. Create them in a controlled dependency-resolution stage; never invent their contents. The official repo requires PyTorch >=2.4.0. Its baseline A14B T2V/I2V commands use 1280*720; the TI2V-5B command uses 1280*704 and CPU/offload flags for a documented 24 GB minimum. Capacity claims are baseline guidance, not guarantees for other frame counts, drivers, or colocated workloads.

Open Wan 2.1 and 2.2 code/checkpoints are labeled Apache License 2.0. Preserve notices and satisfy Apache conditions when redistributing the work or derivatives. The Wan team says it claims no rights in generated content, but that is not a warranty of non-infringement or ownership: outputs can resemble protected material, people, brands, or other users' outputs. Audit input rights, publicity/privacy/biometric consent, music/voice licenses, and the laws of the deployment jurisdiction. Commun

…(truncated)