hookbase 2.2.0

Official Python SDK for the Hookbase webhook management platform

Hookbase Python SDK

The official Python SDK for the Hookbase webhook management platform.

Installation

pip install hookbase

Quick Start

from hookbase import Hookbase

client = Hookbase(api_key="whr_your_api_key")

# List webhook sources
sources = client.sources.list()
for source in sources.data:
    print(f"{source.name} ({source.provider})")

# Send an outbound webhook
result = client.outbound.messages.send(
    "app_123",
    event_type="order.created",
    payload={"orderId": "456", "amount": 99.99},
)
print(f"Sent: {result.event_id}")

Async Support

from hookbase import AsyncHookbase

async with AsyncHookbase(api_key="whr_...") as client:
    sources = await client.sources.list()

API Reference

Client Initialization

from hookbase import Hookbase

client = Hookbase(
    api_key="whr_...",           # Required
    base_url="https://...",      # Default: https://api.hookbase.app
    timeout=30.0,                # Seconds (default: 30)
    max_retries=3,               # Default: 3
    debug=False,                 # Log requests
)

Inbound Webhooks

# Sources
client.sources.list(search="github", provider="github")
client.sources.create({"name": "GitHub", "slug": "github", "provider": "github"})
client.sources.get("src_id")
client.sources.update("src_id", {"name": "Updated"})
client.sources.delete("src_id")
client.sources.rotate_secret("src_id")

# Destinations
client.destinations.list()
client.destinations.create({"name": "Backend", "url": "https://..."})
client.destinations.test("dst_id")

# Routes
client.routes.create({"name": "Route", "sourceId": "src_id", "destinationId": "dst_id"})
client.routes.get_circuit_status("route_id")
client.routes.reset_circuit("route_id")

# Events & Deliveries
client.events.list(source_id="src_id", from_date="2024-01-01")
client.events.get("evt_id")  # Includes payload and deliveries
client.deliveries.replay("del_id")
client.deliveries.bulk_replay(["del_1", "del_2"])

# Transforms & Filters
client.transforms.create({"name": "Extract", "transformType": "jsonata", "code": "$.data"})
client.transforms.test("txf_id", payload={"data": {"key": "value"}})
client.filters.test([{"field": "$.type", "operator": "eq", "value": "order"}], payload={...})

# Schemas
client.schemas.create({"name": "OrderSchema", "jsonSchema": {"type": "object", ...}})

Outbound Webhooks

# Applications
client.outbound.applications.create({"name": "Acme", "uid": "cust_123"})
client.outbound.applications.get_by_external_id("cust_123")

# Endpoints
client.outbound.endpoints.create("app_id", {"url": "https://..."})
client.outbound.endpoints.rotate_secret("ep_id", grace_period=3600)

# Event Types & Subscriptions
client.outbound.event_types.create({"name": "order.created", "category": "orders"})
client.outbound.subscriptions.create({"endpointId": "ep_id", "eventTypeId": "et_id"})
client.outbound.subscriptions.bulk_create("ep_id", ["et_1", "et_2"])

# Send Events
client.outbound.messages.send("app_id", event_type="order.created", payload={...})

# Message Log
client.outbound.message_log.list(application_id="app_id")
client.outbound.message_log.list_attempts("msg_id")
client.outbound.message_log.stats()

# Dead Letter Queue
client.outbound.dlq.list()
client.outbound.dlq.stats()
client.outbound.dlq.retry("dlq_id")
client.outbound.dlq.retry_bulk(["dlq_1", "dlq_2"])

Admin

# Organizations
client.organizations.list()
client.organizations.list_members("org_id")
client.organizations.invite("org_id", "[e-mail]", "member")

# API Keys
client.api_keys.create({"name": "CI Key", "scopes": ["read", "write"]})

# Analytics
client.analytics.dashboard(range="30d")

# Cron Jobs & Tunnels
client.cron_jobs.list()
client.tunnels.list()

Pagination

# Offset-based (sources, destinations, routes, etc.)
page = client.sources.list(page_size=10)
print(f"Total: {page.total}, Page: {page.page}")

# Auto-paginate through all items
for source in page.auto_paging_iter():
    print(source.name)

# Manual pagination
while page.has_more:
    page = page.next_page()

# Cursor-based (applications, endpoints, subscriptions, etc.)
page = client.outbound.applications.list(limit=50)
for app in page.auto_paging_iter():
    print(app.name)

Webhook Verification

from hookbase import Webhook, WebhookVerificationError

wh = Webhook("whsec_your_secret")

try:
    payload = wh.verify(request_body, request_headers)
    print(f"Verified: {payload}")
except WebhookVerificationError as e:
    print(f"Invalid: {e}")

# Generate test headers
headers = wh.generate_test_headers('{"test": true}')

Error Handling

from hookbase import (
    HookbaseError,          # Base error
    APIError,               # API returned an error
    AuthenticationError,    # 401
    ForbiddenError,         # 403
    NotFoundError,          # 404
    ValidationError,        # 400/422
    ConflictError,          # 409
    RateLimitError,         # 429 (has retry_after)
    TimeoutError,           # Request timed out
    NetworkError,           # Connection failed
    WebhookVerificationError,
)

try:
    client.sources.get("src_id")
except NotFoundError:
    print("Not found")
except RateLimitError as e:
    print(f"Retry after {e.retry_after}s")
except APIError as e:
    print(f"{e.status_code}: {e} (request_id={e.request_id})")

Pydantic Models

All responses are typed Pydantic models with camelCase alias support:

from hookbase.models import Source, CreateSourceParams

# Use Pydantic models for type safety
params = CreateSourceParams(name="GitHub", provider="github")
source = client.sources.create(params)

# Or use plain dicts
source = client.sources.create({"name": "GitHub", "provider": "github"})

Development

pip install -e ".[dev]"
pytest tests/ -v
mypy src/hookbase
ruff check src/ tests/

License

MIT