Webhooks

Overview

The SnipeRoute Python SDK provides utilities for:

Verifying webhook signatures - Ensure requests come from SnipeRoute
Managing webhooks - Create, update, delete webhooks via API
Type-safe models - Pydantic models for webhook payloads

Signature Verification

Quick Start

from fastapi import FastAPI, Request, HTTPException
from sniperoute import verify_webhook_signature, WebhookVerificationError
 
app = FastAPI()
 
WEBHOOK_SECRET = "your_hex_encoded_secret"
 
@app.post("/webhooks/sniperoute")
async def handle_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("sr-signature")
    timestamp = request.headers.get("sr-timestamp")
 
    try:
        verify_webhook_signature(
            payload=body,
            signature_header=signature,
            timestamp=timestamp,
            secret=WEBHOOK_SECRET,
        )
    except WebhookVerificationError as e:
        raise HTTPException(status_code=401, detail=e.message)
 
    event = await request.json()
 
    # Handle the event
    if event["type"] == "intent.filled":
        print(f"Order filled: {event['data']['intent_id']}")
 
    return {"status": "received"}

verify_webhook_signature()

def verify_webhook_signature(
    payload: str | bytes,
    signature_header: str,
    timestamp: str | int,
    secret: str,
    tolerance: int = 300,
) -> bool

Parameters:

Parameter	Type	Description
`payload`	str \| bytes	Raw request body
`signature_header`	str	Value of `sr-signature` header
`timestamp`	str \| int	Value of `sr-timestamp` header
`secret`	str	Hex-encoded webhook signing secret
`tolerance`	int	Max age of webhook in seconds (default: 300)

Returns: True if signature is valid

Raises: WebhookVerificationError with specific reason

WebhookSigner Class

For more control over verification:

from sniperoute import WebhookSigner, WebhookVerificationError
 
signer = WebhookSigner(
    secret="your_hex_encoded_secret",
    tolerance=300  # 5 minutes
)
 
# Verify signature
try:
    signer.verify(payload, signature_header, timestamp)
except WebhookVerificationError as e:
    print(f"Failed: {e.message} (reason: {e.reason})")
 
# Compute signature (for testing)
signature = signer.compute_signature(payload, timestamp)

Error Handling

WebhookVerificationError provides detailed error information:

from sniperoute import WebhookVerificationError
 
try:
    verify_webhook_signature(payload, signature, timestamp, secret)
except WebhookVerificationError as e:
    print(f"Message: {e.message}")
    print(f"Reason: {e.reason}")

Error Reasons:

Reason	Description
`missing_signature`	`sr-signature` header not provided
`missing_timestamp`	`sr-timestamp` header not provided
`missing_event_id`	`sr-event-id` header not provided
`invalid_timestamp`	Timestamp is not a valid integer
`timestamp_expired`	Timestamp outside tolerance window
`invalid_signature_format`	Signature doesn't start with `v1=`
`signature_mismatch`	Computed signature doesn't match

Webhook Management

Use the SnipeRouteClient to manage webhooks programmatically.

Create Webhook

from sniperoute import SnipeRouteClient, WebhookCreate
 
async with SnipeRouteClient() as client:
    webhook = WebhookCreate(
        url="https://api.yourapp.com/webhooks/sniperoute",
        events=["intent.filled", "intent.rejected", "intent.partially_filled"],
        description="Production webhook"
    )
 
    response = await client.create_webhook(webhook)
 
    print(f"Webhook ID: {response.id}")
    print(f"Secret: {response.secret}")  # Save this! Only shown once

⚠️

The webhook secret is only returned when creating a webhook. Store it securely immediately.

List Webhooks

webhooks = await client.list_webhooks()
 
for webhook in webhooks:
    print(f"{webhook.id}: {webhook.url}")
    print(f"  Events: {', '.join(webhook.events)}")
    print(f"  Active: {webhook.is_active}")

Update Webhook

from sniperoute import WebhookUpdate
 
# Disable webhook
update = WebhookUpdate(is_active=False)
updated = await client.update_webhook("wh_abc123", update)
 
# Change events
update = WebhookUpdate(
    events=["intent.filled", "intent.canceled"]
)
updated = await client.update_webhook("wh_abc123", update)
 
# Update URL
update = WebhookUpdate(
    url="https://new-api.yourapp.com/webhooks/sniperoute"
)
updated = await client.update_webhook("wh_abc123", update)

Delete Webhook

await client.delete_webhook("wh_abc123")

Rotate Secret

response = await client.rotate_webhook_secret("wh_abc123")
print(f"New secret: {response.secret}")
 
# Update your webhook handler with the new secret immediately!

⚠️

Rotating the secret immediately invalidates the old one. Update your webhook handler before rotating.

View Delivery History

deliveries = await client.list_webhook_deliveries(
    "wh_abc123",
    limit=20,
    offset=0
)
 
for delivery in deliveries:
    print(f"{delivery.event_type}: {delivery.status}")
    print(f"  HTTP Status: {delivery.http_status}")
    if delivery.error_message:
        print(f"  Error: {delivery.error_message}")
    print(f"  Attempts: {delivery.attempt_count}")

Send Test Event

response = await client.send_test_webhook("wh_abc123")
print(f"Test event ID: {response.event_id}")
print(f"Delivery ID: {response.delivery_id}")

Models

WebhookCreate

from sniperoute import WebhookCreate
 
webhook = WebhookCreate(
    url="https://...",           # Required: HTTPS URL
    events=["intent.filled"],    # Required: Event types to subscribe
    description="My webhook"     # Optional: Description
)

WebhookUpdate

from sniperoute import WebhookUpdate
 
update = WebhookUpdate(
    url="https://...",           # Optional: New URL
    events=["intent.filled"],    # Optional: New event types
    description="Updated",       # Optional: New description
    is_active=True               # Optional: Enable/disable
)

WebhookResponse

webhook: WebhookResponse
 
webhook.id           # Webhook ID
webhook.url          # Webhook URL
webhook.events       # List of subscribed events
webhook.description  # Description
webhook.is_active    # Whether active
webhook.created_at   # Creation timestamp
webhook.updated_at   # Last update timestamp

WebhookCreateResponse

Extends WebhookResponse with:

response.secret  # HMAC signing secret (shown only once)

WebhookDelivery

delivery: WebhookDelivery
 
delivery.id             # Delivery ID
delivery.event_type     # Event type delivered
delivery.event_id       # Unique event ID
delivery.payload        # Event payload dict
delivery.status         # Delivery status
delivery.http_status    # HTTP response status
delivery.response_body  # Response body (truncated)
delivery.error_message  # Error if failed
delivery.attempt_count  # Number of attempts
delivery.next_retry_at  # Next retry timestamp
delivery.created_at     # Creation timestamp
delivery.delivered_at   # Success timestamp

WebhookEvent

from sniperoute import WebhookEvent
 
event = WebhookEvent(
    type="intent.filled",
    id="evt_abc123",
    created_at=datetime.now(),
    data={"intent_id": "trade_001", ...}
)

WebhookEventType

Enum of all supported event types:

from sniperoute import WebhookEventType
 
WebhookEventType.INTENT_CREATED
WebhookEventType.INTENT_PENDING
WebhookEventType.INTENT_QUEUED
WebhookEventType.INTENT_COMPLETED
WebhookEventType.INTENT_FILLED
WebhookEventType.INTENT_PARTIALLY_FILLED
WebhookEventType.INTENT_CANCELED
WebhookEventType.INTENT_REJECTED
WebhookEventType.INTENT_FAILED
 
WebhookEventType.ORDER_SUBMITTED
WebhookEventType.ORDER_FILLED
WebhookEventType.ORDER_PARTIAL_FILLED
WebhookEventType.ORDER_CANCELED
WebhookEventType.ORDER_REJECTED
WebhookEventType.ORDER_EXPIRED
 
WebhookEventType.TEST_PING

Complete Example

import asyncio
from fastapi import FastAPI, Request, HTTPException
from sniperoute import (
    SnipeRouteClient,
    WebhookCreate,
    verify_webhook_signature,
    WebhookVerificationError,
    WebhookEventType,
)
 
app = FastAPI()
 
# Store this securely (e.g., environment variable)
WEBHOOK_SECRET = "your_hex_encoded_secret"
 
@app.post("/webhooks/sniperoute")
async def handle_webhook(request: Request):
    # 1. Get headers and body
    body = await request.body()
    signature = request.headers.get("sr-signature")
    timestamp = request.headers.get("sr-timestamp")
    event_id = request.headers.get("sr-event-id")
 
    # 2. Verify signature
    try:
        verify_webhook_signature(
            payload=body,
            signature_header=signature,
            timestamp=timestamp,
            secret=WEBHOOK_SECRET,
        )
    except WebhookVerificationError as e:
        print(f"Verification failed: {e.reason}")
        raise HTTPException(status_code=401, detail=e.message)
 
    # 3. Parse event
    event = await request.json()
    event_type = event["type"]
 
    # 4. Handle event (use event_id for idempotency)
    if event_type == WebhookEventType.INTENT_FILLED.value:
        data = event["data"]
        print(f"Order filled: {data['intent_id']}")
        print(f"  Symbol: {data['symbol']}")
        print(f"  Quantity: {data['quantity']}")
        # Update database, notify user, etc.
 
    elif event_type == WebhookEventType.INTENT_REJECTED.value:
        data = event["data"]
        print(f"Order rejected: {data['intent_id']}")
        print(f"  Error: {data.get('error', {}).get('message')}")
        # Handle rejection
 
    elif event_type == WebhookEventType.INTENT_PARTIALLY_FILLED.value:
        data = event["data"]
        print(f"Order partially filled: {data['intent_id']}")
        # Track partial fill
 
    return {"status": "received"}
 
 
async def setup_webhook():
    """Example: Create a webhook programmatically."""
    async with SnipeRouteClient() as client:
        webhook = WebhookCreate(
            url="https://api.yourapp.com/webhooks/sniperoute",
            events=[
                WebhookEventType.INTENT_FILLED.value,
                WebhookEventType.INTENT_REJECTED.value,
                WebhookEventType.INTENT_PARTIALLY_FILLED.value,
            ],
            description="Production webhook"
        )
 
        response = await client.create_webhook(webhook)
        print(f"Created webhook: {response.id}")
        print(f"Secret: {response.secret}")  # Store securely!
 
        # Send a test event
        test = await client.send_test_webhook(response.id)
        print(f"Test event sent: {test.event_id}")
 
 
if __name__ == "__main__":
    asyncio.run(setup_webhook())

Next Steps

Webhook Events

View all webhook event types and payloads

Signature Verification

Deep dive into signature verification

Error Handling

Exception hierarchy and handling

Client Reference

Full client API reference

Client Reference Models