Skip to main content
SDKs & Libraries

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 immediately

Important: 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}")

Related