Webhooks

import os
from spine import SpineClient, WebhookSignatureError

client = SpineClient()  # SPINE_API_KEY picked up from env

def handle_spine_webhook(raw_body: bytes, sig_header: str) -> dict:
    try:
        event = client.webhooks.construct_event(
            payload=raw_body,
            sig_header=sig_header,
            secret=os.environ["SPINE_WEBHOOK_SECRET"],
        )
    except WebhookSignatureError:
        return {"ok": False}, 400

    if event.type == "run.completed":
        run = event.data["object"]
        # persist run["run_id"] etc.
    return {"ok": True}, 200

from fastapi import FastAPI, Header, HTTPException, Request
from spine import SpineClient, WebhookSignatureError

app = FastAPI()
client = SpineClient()

@app.post("/webhooks/spine")
async def spine_webhook(
    request: Request,
    spine_signature: str = Header(default=""),
):
    raw = await request.body()
    try:
        event = client.webhooks.construct_event(
            payload=raw,
            sig_header=spine_signature,
            secret=os.environ["SPINE_WEBHOOK_SECRET"],
        )
    except WebhookSignatureError as exc:
        raise HTTPException(status_code=400, detail=str(exc))

    # Dedupe on event.id, then dispatch
    return {"ok": True}

from flask import Flask, request
from spine import SpineClient, WebhookSignatureError

app = Flask(__name__)
client = SpineClient()

@app.post("/webhooks/spine")
def spine_webhook():
    try:
        event = client.webhooks.construct_event(
            payload=request.get_data(),                      # raw bytes
            sig_header=request.headers.get("Spine-Signature", ""),
            secret=os.environ["SPINE_WEBHOOK_SECRET"],
        )
    except WebhookSignatureError:
        return {"ok": False}, 400

    # Dedupe on event.id, then dispatch
    return {"ok": True}, 200

from spine import AsyncSpineClient

async with AsyncSpineClient() as client:
    event = client.webhooks.construct_event(...)