Webhooks

endpoint = await client.webhooks.create({
    "url": "https://your-app.com/webhooks/keystone",
    "description": "Production webhook",
    "events": ["settlement.state.*", "settlement.state.finalized"],
})

# IMPORTANT: store the secret securely - only returned on creation
print(endpoint.secret)

result = await client.webhooks.list()
endpoint = await client.webhooks.get("endpoint-uuid")

await client.webhooks.update("endpoint-uuid", {"events": ["*"], "is_active": True})
await client.webhooks.delete("endpoint-uuid")

result = await client.webhooks.test("endpoint-uuid")
print(f"Success: {result.success}, Status: {result.status_code}, Time: {result.duration_ms}ms")

result = await client.webhooks.rotate_secret("endpoint-uuid")
print(f"New: {result.new_secret}")
print(f"Old valid until: {result.previous_secret_valid_until}")

result = await client.webhooks.list_deliveries("endpoint-uuid")

for d in result.items:
    status = "OK" if d.success else "FAILED"
    print(f"{d.event}: {status} ({d.response_status}) - {d.duration_ms}ms")

from keystoneos.utils.webhook_verify import verify_signature

# FastAPI
@app.post("/webhooks/keystone")
async def handle_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("x-keystone-signature", "")

    if not verify_signature(body, WEBHOOK_SECRET, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = await request.json()
    match event["event"]:
        case "settlement.state.finalized":
            print(f"Finalized: {event['data']['settlement_id']}")
        case name if name.startswith("settlement.state."):
            print(f"{event['data']['settlement_id']}: {event['data']['state']}")

    return {"status": "ok"}