Task Callbacks

from beam import function


@function(callback_url="https://your-service.example/task-result")
def handler(x):
    return {"result": x}


if __name__ == "__main__":
    handler.remote(x=10)

{"data": {"result": 10}}

from beam import endpoint


@endpoint()
def handler(callback_url=None):
    return {"ok": True}

curl --fail-with-body -X POST "$BEAM_ENDPOINT_URL" \
  -H "Authorization: Bearer $BEAM_TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"callback_url":"https://your-service.example/task-result"}'

import base64
import hashlib
import hmac
import time


def verify_callback(raw_body: bytes, secret: str, timestamp: str, signature: str) -> bool:
    try:
        signed_at = int(timestamp)
    except ValueError:
        return False
    # Choose a tolerance that accommodates delivery delays and clock skew.
    if abs(time.time() - signed_at) > 300:
        return False
    encoded = base64.b64encode(raw_body).decode()
    expected = hmac.new(
        secret.encode(), f"{encoded}:{timestamp}".encode(), hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)