Kling motion control job done: a signed Python webhook receiver

To get a Kling 3.0 motion-control result without polling, submit the job with mode: "webhook" and a public HTTPS webhook_url, then verify each delivery by HMAC SHA-256 over <timestamp>.<raw_body> using your workspace signing secret. Sume sends only terminal events: job.completed, job.failed and job.canceled. A receiver must refuse an empty secret, check the timestamp window, and compare signatures in constant time.

The route is POST /v1/kling/3.0/motion-control: a still (or ready avatar) plus a motion_video_url whose movement drives the output, and a declared duration_seconds from 1 to 30 (Models, read 2026-10-06). The signing scheme and headers come from Webhooks, read the same day.

How do you submit with a webhook?

Send the same body you would for polling, add the webhook fields and an idempotency key. The still and the motion video must be reachable public HTTPS URLs.

curl -X POST https://api.sume.com/v1/kling/3.0/motion-control \
  -H "Authorization: Bearer $SUME_API_KEY" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: kling-hook-001" \
  -d '{
    "image_url": "https://example.com/mascot.png",
    "motion_video_url": "https://example.com/move.mp4",
    "duration_seconds": 10,
    "character_orientation": "video",
    "mode": "webhook",
    "webhook_url": "https://hooks.example.com/sume"
  }'

How do you verify the delivery?

Read the raw body bytes before any JSON parsing, because the signature covers them exactly. The header can carry more than one sume-v1= entry during a secret rotation, so accept the delivery if any entry matches. Reject anything outside five minutes, a reasonable replay window from the docs.

import hashlib, hmac, os, time
from http.server import BaseHTTPRequestHandler, HTTPServer

SECRET = os.environ.get("SUME_COM_WEBHOOK_SIGNING_SECRET", "")

def verify(raw, ts, header, secret, tolerance=300, now=None):
    if not secret:
        return False
    try:
        stamp = int(ts)
    except ValueError:
        return False
    if abs((now or time.time()) - stamp) > tolerance:
        return False
    mac = hmac.new(secret.encode(), ts.encode() + b"." + raw, hashlib.sha256)
    want = "sume-v1=" + mac.hexdigest()
    return any(hmac.compare_digest(p.strip(), want) for p in header.split(","))

class Hook(BaseHTTPRequestHandler):
    def do_POST(self):
        raw = self.rfile.read(int(self.headers.get("Content-Length", "0")))
        ok = verify(raw, self.headers.get("x-sume-webhook-timestamp", ""),
                    self.headers.get("x-sume-webhook-signature", ""), SECRET)
        self.send_response(204 if ok else 401)
        self.end_headers()
        if ok:
            print(raw.decode())

if __name__ == "__main__":
    HTTPServer(("", 8080), Hook).serve_forever()

What should the handler do next?

Answer well inside the 10 second per-attempt timeout and do the slow work afterwards: store the job_id, fetch GET /v1/jobs/:id/result, and copy the clip to your own storage. A webhook is a notification, so keep a poll fallback for jobs that never call back, as Jobs and results recommends. Make the handler idempotent on job_id, since a delivery can arrive more than once.

Check the cost at submit: the price is ceil(duration_seconds) times the motion-control list rate times 1.25 with the rate at $0.126 per output second before margin, so a 10-second declared duration is about $1.58 ($1.26 times 1.25, ceil to the cent: $1.58). Whether the clip has sound is your choice with keep_original_sound, which defaults to true, as covered in the silent clip post. For orientation choices, see the orientation guide.

How do you test the receiver before real traffic?

Sign a sample body yourself. Compute HMAC-SHA256 over the timestamp, a dot and the raw bytes, hex-encode it, and send it with the sume-v1= prefix and a current timestamp. Then flip a character in the body and confirm you get a rejection, set the timestamp ten minutes old and confirm another, and start the process with an empty secret and confirm it refuses to verify anything at all.

Respond quickly and do the work after. Acknowledge with a 2xx as soon as the signature checks out, then enqueue the result handling. Treat job.completed, job.failed and job.canceled as the three outcomes, and make handlers idempotent, because a delivery may arrive more than once. During secret rotation the header can carry several sume-v1 entries, and the verifier should accept the request if any one matches.

What happens when your endpoint is down

The webhooks page says Sume makes up to 10 delivery attempts in total, with a fixed delay between them (30 seconds by default, not exponential backoff), and each attempt times out at 10 seconds. A slow endpoint spends that budget, and an error or non-2xx response is retried. After ten refused attempts you have a failed delivery, but the job still reached its real terminal state, so the result is waiting at the status and result URLs.

That is the reason to keep a poll fallback and to key your handler on job_id: the same event can be delivered more than once, and an event can also never arrive. A nightly sweep that lists your open job ids and fetches any that Sume reports as terminal covers both cases without any change to the verifier.