Webhook endpoints can also be managed in the KeyStone Dashboard under Settings > Webhooks - no code required.
create
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)
| Field | Type | Required | Default | Description |
|---|
url | str | Yes | - | HTTPS endpoint URL |
description | str | No | - | Human-readable label |
events | list[str] | No | ["*"] | Event patterns |
secret | str | No | auto-generated | Custom signing secret |
list / get / update / delete
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")
test
result = await client.webhooks.test("endpoint-uuid")
print(f"Success: {result.success}, Status: {result.status_code}, Time: {result.duration_ms}ms")
rotate_secret
Previous secret stays valid for 24 hours.
result = await client.webhooks.rotate_secret("endpoint-uuid")
print(f"New: {result.new_secret}")
print(f"Old valid until: {result.previous_secret_valid_until}")
list_deliveries
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")
verify_signature
Synchronous, local operation - no API call.
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.*":
print(f"{event['data']['settlement_id']}: {event['data']['to_state']}")
case "settlement.state.finalized":
print(f"Finalized: {event['data']['settlement_id']}")
return {"status": "ok"}
Use await request.body() (raw bytes) for signature verification, not await request.json(). Parsing can change whitespace and break the signature.