Python SDK
The official XRNotify Python SDK provides a synchronous and async-compatible interface for managing webhooks, querying deliveries, and verifying signatures. Full Pydantic models are included for type-safe event handling.
Installation
pip install xrnotify
Python
3.8+
HTTP
httpx
Types
pydantic
Initialization
Import and instantiate the client with your API key from an environment variable.
import os from xrnotify import XRNotify client = XRNotify(api_key=os.environ["XRNOTIFY_API_KEY"]) # Optional: use the test environment # client = XRNotify( # api_key=os.environ["XRNOTIFY_TEST_KEY"], # environment="test" # )
Test environment: Pass environment="test" to use test API keys. Synthetic events will be delivered without consuming real XRPL data.
Creating a webhook
Register a webhook endpoint. The secret attribute is only present on the creation response.
webhook = client.webhooks.create(
url="https://yourapp.com/webhooks/xrpl",
event_types=["payment.xrp", "nft.minted"],
account_filters=["rPT1Sjq2YGrBMTttX4GZHjKu9dyfzbpAYe"],
description="Payment processor"
)
print(webhook.id) # wh_abc123
print(webhook.secret) # Only available on creation — store immediatelyImportant: The secret field is returned once at creation. It cannot be retrieved again. Store it in your secrets manager immediately.
Listing and updating webhooks
List all webhooks in your account and update existing ones. Only the keyword arguments you pass are changed.
# List webhooks
result = client.webhooks.list(limit=10)
for wh in result.data:
print(wh.id, wh.url, wh.is_active)
# Paginate
if result.has_more:
next_page = client.webhooks.list(
limit=10,
starting_after=result.data[-1].id
)
# Pause a webhook
client.webhooks.update("wh_abc123", is_active=False)
# Update subscribed event types
client.webhooks.update("wh_abc123", event_types=["payment.*", "nft.*"])
# Delete a webhook
client.webhooks.delete("wh_abc123")Listing deliveries
Query delivery history filtered by webhook, status, and time range. Pass a timezone-aware datetime for the since parameter.
from datetime import datetime, timezone
result = client.deliveries.list(
webhook_id="wh_abc123",
status="failed",
since=datetime(2024, 1, 1, tzinfo=timezone.utc)
)
for delivery in result.data:
print(
delivery.id,
delivery.event_type,
delivery.status_code,
delivery.last_error
)Verifying signatures
Use verify_signature from the SDK to validate incoming webhook requests. The function uses hmac.compare_digest internally for constant-time comparison.
import os
from flask import Flask, request
from xrnotify import verify_signature
app = Flask(__name__)
@app.route("/webhooks/xrpl", methods=["POST"])
def handle_webhook():
# Verify before parsing — use request.data (raw bytes)
if not verify_signature(
payload=request.data,
signature=request.headers.get("X-XRNotify-Signature", ""),
secret=os.environ["WEBHOOK_SECRET"]
):
return "Unauthorized", 401
event = request.get_json(force=True)
handle_event(event)
return "OK", 200
def handle_event(event: dict) -> None:
event_type = event.get("event_type")
if event_type == "payment.xrp":
payload = event["payload"]
print(f"XRP payment: {payload['amount_xrp']} XRP")
elif event_type == "nft.minted":
payload = event["payload"]
print(f"NFT minted: {payload['nft_id']}")Raw body required: Pass request.data (bytes), not request.get_json(). The signature is computed over the raw bytes before any JSON parsing.
Pydantic models
The SDK ships Pydantic v2 models for every event type. Use them for validation, serialization, and IDE autocompletion.
from xrnotify.models import (
WebhookEvent,
PaymentXrpPayload,
PaymentIssuedPayload,
NftMintedPayload,
DexOfferFilledPayload,
)
def handle_event(data: dict) -> None:
event = WebhookEvent(**data)
if event.event_type == "payment.xrp":
payload = PaymentXrpPayload(**event.payload)
print(f"Received {payload.amount_xrp} XRP")
print(f"From: {payload.sender} → {payload.receiver}")
elif event.event_type == "nft.minted":
payload = NftMintedPayload(**event.payload)
print(f"NFT {payload.nft_id} minted by {payload.issuer}")
if payload.uri_decoded:
print(f"Metadata: {payload.uri_decoded}")