HookBridge Python SDK

Official HookBridge SDK for Python. Send webhooks with guaranteed delivery, automatic retries, and full observability.

Installation

pip install hookbridge

Quick Start

from hookbridge import HookBridge

client = HookBridge(api_key="hb_live_xxxxxxxxxxxxxxxxxxxx")

# Create an endpoint (do this once, save the endpoint_id)
endpoint = client.create_endpoint(
    url="https://customer.app/webhooks",
    description="Customer webhook receiver"
)
print("Endpoint ID:", endpoint.id)
print("Signing Secret:", endpoint.signing_secret)  # Save this!

# Send webhooks to the endpoint
result = client.send(
    endpoint_id=endpoint.id,
    payload={
        "event": "order.created",
        "order_id": "ord_12345",
        "amount": 99.99
    }
)

print("Message ID:", result.message_id)

Features

• Guaranteed Delivery: Webhooks are stored durably before acknowledgment
• Automatic Retries: Intelligent retry with exponential backoff
• Idempotency: Prevent duplicate webhook sends
• Full Observability: Query logs, metrics, and message status
• Type Safety: Full type hints with mypy support
• Async Support: Both sync and async clients available

Usage

Manage Endpoints

# Create an endpoint
endpoint = client.create_endpoint(
    url="https://customer.app/webhooks",
    description="Production webhook receiver",
    rate_limit_rps=10,  # Optional: rate limit
    burst=20,           # Optional: burst allowance
)

# Get endpoint details
endpoint = client.get_endpoint(endpoint_id)

# List all endpoints
endpoints = client.list_endpoints(limit=50)
for ep in endpoints.endpoints:
    print(ep.id, ep.url)

# Update an endpoint
client.update_endpoint(endpoint_id, description="Updated description")

# Rotate signing secret
new_secret = client.rotate_endpoint_secret(endpoint_id)
print("New secret:", new_secret.signing_secret)

# Delete an endpoint
client.delete_endpoint(endpoint_id)

Send a Webhook

result = client.send(
    endpoint_id="ep_550e8400e29b41d4a716446655440000",
    payload={"event": "user.created", "user_id": "usr_123"},
    headers={"X-Tenant-Id": "tenant_abc"},
    idempotency_key="user-123-created"
)

Check Message Status

message = client.get_message(result.message_id)
print(message.status)  # 'queued', 'succeeded', 'pending_retry', 'failed_permanent'

Query Delivery Logs

from datetime import datetime

logs = client.get_logs(
    status="failed_permanent",
    start_time=datetime(2025, 1, 1),
    limit=100
)

for msg in logs.messages:
    print(msg.message_id, msg.last_error)

Get Metrics

metrics = client.get_metrics("24h")
print(f"Success rate: {metrics.success_rate * 100:.1f}%")
print(f"Average latency: {metrics.avg_latency_ms}ms")

Replay Failed Messages

# Replay a specific message
client.replay(message_id)

# Or replay from the Dead Letter Queue
dlq = client.get_dlq_messages()
for msg in dlq.messages:
    client.replay_from_dlq(msg.message_id)

Manage Retries

# Cancel a pending retry (moves to DLQ)
client.cancel_retry(message_id)

# Trigger immediate retry for a pending message
client.retry_now(message_id)

API Key Management

# List API keys
keys = client.list_api_keys()

# Create a new API key
new_key = client.create_api_key(mode="live", label="Production backend")
print("Save this key:", new_key.key)  # Only shown once!

# Delete an API key
client.delete_api_key("key_abc123")

Inbound Webhooks

# Create an inbound endpoint
inbound = client.create_inbound_endpoint(
    url="https://myapp.com/webhooks/inbound",
    name="Stripe inbound",
    description="Receives Stripe events through HookBridge",
    verify_static_token=True,
    token_header_name="X-Webhook-Token",
    token_value="my-shared-secret",
    signing_enabled=True,
    idempotency_header_names=["X-Idempotency-Key"],
    ingest_response_code=202,
)

print("Inbound endpoint ID:", inbound.id)
print("Send webhooks here:", inbound.ingest_url)   # Save this
print("Secret token:", inbound.secret_token)       # Only shown once

# Inspect and manage the inbound endpoint
details = client.get_inbound_endpoint(inbound.id)
client.pause_inbound_endpoint(inbound.id)
client.resume_inbound_endpoint(inbound.id)

# Update verification settings later if needed
client.update_inbound_endpoint(
    inbound.id,
    verify_hmac=True,
    hmac_header_name="X-Signature",
    hmac_secret="whsec_inbound_secret",
)

Inbound Observability

# List inbound endpoints
inbound_endpoints = client.list_inbound_endpoints(limit=50)

# Query inbound delivery logs
logs = client.get_inbound_logs(
    inbound_endpoint_id="01935abc-def0-7123-4567-890abcdef012",
    limit=50,
)
for entry in logs.entries:
    print(entry.message_id, entry.status, entry.total_delivery_ms)

# Get inbound metrics and time series
metrics = client.get_inbound_metrics(
    "24h",
    inbound_endpoint_id="01935abc-def0-7123-4567-890abcdef012",
)
timeseries = client.get_inbound_timeseries_metrics(
    "24h",
    inbound_endpoint_id="01935abc-def0-7123-4567-890abcdef012",
)

# Review rejected inbound requests
rejections = client.list_inbound_rejections(
    inbound_endpoint_id="01935abc-def0-7123-4567-890abcdef012",
    limit=25,
)
for rejection in rejections.entries:
    print(rejection.reason_code, rejection.reason_detail)

Async Usage

from hookbridge import AsyncHookBridge

async with AsyncHookBridge(api_key="hb_live_xxx") as client:
    # Create endpoint
    endpoint = await client.create_endpoint(
        url="https://customer.app/webhooks"
    )

    # Send webhook
    result = await client.send(
        endpoint_id=endpoint.id,
        payload={"event": "order.created"}
    )

Error Handling

from hookbridge import (
    HookBridge,
    AuthenticationError,
    ValidationError,
    RateLimitError,
    NotFoundError,
    IdempotencyError
)

try:
    client.send(...)
except AuthenticationError:
    print("Invalid API key")
except ValidationError as e:
    print(f"Invalid request: {e}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except IdempotencyError:
    print("Duplicate request with different payload")

Webhook Delivery

When HookBridge delivers your webhook, it includes these headers:

• X-Webhook-Signature: HMAC-SHA256 signature for verification
• X-Webhook-Id: Message ID for tracking
• X-Webhook-Timestamp: Unix timestamp of the send request
• Any custom headers you specified

Retry Behavior

• Fast retries (for 429 responses): 30s, 60s, 120s, 240s, 300s
• Slow retries (for other errors): 30m, 2h, 6h, 12h, 24h, 48h, 72h, 96h
• Maximum 8 total attempts before moving to the Dead Letter Queue

Context Manager Support

# Sync
with HookBridge(api_key="...") as client:
    client.send(...)

# Async
async with AsyncHookBridge(api_key="...") as client:
    await client.send(...)

Requirements

• Python 3.9 or later
• httpx

License

MIT