hookbridge 1.8.0

Official HookBridge SDK for Python

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