Webhooks - KeyStone OS

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 (max 50, each max 100 characters)
`secret`	`str`	No	auto-generated	Custom signing secret

An environment holds at most 50 endpoints. See webhook limits.

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.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"}

Use await request.body() (raw bytes) for signature verification, not await request.json(). Parsing can change whitespace and break the signature.