hookbase

The Hookbase Python SDK provides a typed client for sending webhooks to your customers, with a synchronous Hookbase client and an AsyncHookbase client that mirrors it. It requires Python 3.9 or newer and depends on httpx and pydantic (v2). Responses are Pydantic models, and the package ships type hints (py.typed).

Source: github.com/HookbaseApp/hookbase-python

Installation

pip install hookbase

To verify Ed25519 (v1a) signatures with an endpoint's whpk_ public key, install the ed25519 extra, which adds cryptography:

pip install 'hookbase[ed25519]'

This page describes version 4.1.0.

Quick Start

import os
 
from hookbase import Hookbase
 
client = Hookbase(api_key=os.environ["HOOKBASE_API_KEY"])
 
# An application represents one of your customers
app = client.outbound.applications.create({"name": "Acme Corp", "externalId": "customer_123"})
 
# The endpoint the customer receives webhooks on
endpoint = client.outbound.endpoints.create(app.id, {"url": "https://example.com/webhooks"})
 
# An event type, and a subscription from the endpoint to it
order_created = client.outbound.event_types.create({"name": "order.created"})
client.outbound.subscriptions.create({
    "endpointId": endpoint.id,
    "eventTypeId": order_created.id,
})
 
# Send a webhook
result = client.outbound.messages.send(
    app.id,
    event_type="order.created",
    payload={"orderId": "ord_123", "amount": 99.99},
)
print(f"{result.event_id} queued to {result.messages_queued} endpoint(s)")

Client Configuration

import os
 
from hookbase import Hookbase
 
# Defaults: https://api.hookbase.app, 30 s timeout, 3 retries
client = Hookbase(
    api_key=os.environ["HOOKBASE_API_KEY"],
    base_url="https://api.hookbase.app",  # Optional
    timeout=30.0,  # Optional: seconds, an httpx.Timeout, or None for no timeout
    max_retries=3,  # Optional
)
 
# The client holds a connection pool: close it when done, or use it as a context manager
with Hookbase(api_key=os.environ["HOOKBASE_API_KEY"]) as client:
    apps = client.outbound.applications.list()

Create one client per API key and share it. Outbound resources live under client.outbound (applications, endpoints, event_types, subscriptions, messages, message_log, portal_tokens, dlq). Inbound and admin resources (client.sources, client.events, client.api_keys ...) are on the client itself.

Request bodies are either a params model or a plain dict. A params model takes snake_case fields (CreateApplicationParams(name="Acme", external_id="customer_123")), and a field left at None is not sent. A dict is sent as written, so its keys must be the API's camelCase names ({"externalId": "customer_123"}). Responses are always models with snake_case attributes (app.external_id).

Async Client

AsyncHookbase has the same resources and arguments, with every method a coroutine:

import asyncio
import os
 
from hookbase import AsyncHookbase
 
 
async def main() -> None:
    async with AsyncHookbase(api_key=os.environ["HOOKBASE_API_KEY"]) as client:
        result = await client.outbound.messages.send(
            "519c05c7-ada5-5d3e-9b80-2c0dc9743117",
            event_type="order.created",
            payload={"orderId": "ord_123"},
        )
        print(result.event_id)
 
 
asyncio.run(main())

Custom HTTP Client

Pass your own httpx.Client (or httpx.AsyncClient for AsyncHookbase) to set proxies, TLS settings, transports or connection limits. The SDK never modifies it. Every request goes to an absolute URL built from base_url, with the SDK's Authorization and User-Agent headers (an auth= set on your client is not used for Hookbase requests). If you pass timeout to Hookbase, it applies to each request; if you leave it out, your client's own timeout applies. close() and the context manager leave a client you supplied open, so close it yourself.

import os
 
import httpx
 
from hookbase import Hookbase
 
http = httpx.Client(
    verify="/etc/ssl/corp-ca.pem",
    timeout=10.0,
    limits=httpx.Limits(max_connections=20),
)
# No timeout argument, so the client's 10 s timeout applies
client = Hookbase(api_key=os.environ["HOOKBASE_API_KEY"], http_client=http)
 
# ...
 
client.close()
http.close()

Applications

Applications represent your customers. Each customer should have one application.

Create Application

from hookbase.models import CreateApplicationParams
 
app = client.outbound.applications.create(CreateApplicationParams(
    name="Acme Corp",
    external_id="customer_123",  # Your internal customer ID
    metadata={"plan": "enterprise"},
))

Get or Create Application

# Creates the application, or updates the one with the same external ID.
# Returns a tuple: the application, and whether it was created.
app, created = client.outbound.applications.upsert({
    "name": "Acme Corp",
    "externalId": "customer_123",
})

upsert needs an external ID and raises ValueError before sending the request when there isn't one.

List Applications

# One page
page = client.outbound.applications.list(limit=50)
for app in page.data:
    print(app.name)
 
# Every application, fetched page by page as you iterate
for app in client.outbound.applications.list().auto_paging_iter():
    print(app.name)

Get by External ID

app = client.outbound.applications.get_by_external_id("customer_123")

Event Types

Event types define the webhook events your customers can subscribe to.

Create Event Type

from hookbase.models import CreateEventTypeParams
 
event_type = client.outbound.event_types.create(CreateEventTypeParams(
    name="order.created",
    display_name="Order Created",
    category="orders",
    schema={"type": "object", "required": ["orderId"]},
    example_payload={"orderId": "ord_123"},
))

The schema and example payload can be a dict or JSON text. The SDK sends them as the JSON strings the API expects.

List Event Types

for event_type in client.outbound.event_types.list(category="orders").auto_paging_iter():
    print(event_type.name)

Subscriptions

# Subscribe an endpoint to an event type
sub = client.outbound.subscriptions.create({
    "endpointId": "e25aaf14-ecb3-5f7c-9597-225979d06ed9",
    "eventTypeId": "d14e4a97-4566-51bb-8130-baeae1cbcef8",
})
 
# Several at once
subs = client.outbound.subscriptions.bulk_create("e25aaf14-ecb3-5f7c-9597-225979d06ed9", ["c448e8e1-8501-53a3-be4d-a6278a863484", "b371f66f-f84a-5ee6-ae1d-82ceb50672ab"])

bulk_create returns only the subscriptions it created. Event types the endpoint was already subscribed to are skipped.

Messages

Messages are the webhooks you send to your customers. Sending is on client.outbound.messages. Reading what was sent, and replaying it, is on client.outbound.message_log.

Send Message

result = client.outbound.messages.send(
    "519c05c7-ada5-5d3e-9b80-2c0dc9743117",
    event_type="order.created",
    payload={"orderId": "ord_123"},
    labels={"region": "eu"},
    idempotency_key="order-ord_123-created",  # Optional
)
 
print(f"{result.event_id} queued to {result.messages_queued} endpoint(s)")

Every send carries an idempotency key: the one you pass, or a UUID generated for the call. The API deduplicates on it, so a send retried after a timeout is delivered once. Pass endpoint_ids=[...] to deliver to some of the application's endpoints only, and priority (0, critical, to 4, bulk) to change its queue priority.

List Messages and Attempts

for message in client.outbound.message_log.list(
    application_id="519c05c7-ada5-5d3e-9b80-2c0dc9743117", status="failed"
).auto_paging_iter():
    attempts = client.outbound.message_log.list_attempts(message.id)
    print(message.id, message.status, len(attempts))

Retry Failed Message

# Replays the message (creates a new message)
replay = client.outbound.message_log.replay("5c588ab6-d167-5c12-8735-fd1df20421b0")
print(replay.new_message_id)

Messages that ran out of retries are also listed in the dead-letter queue, client.outbound.dlq (list, stats, retry, retry_bulk, delete, bulk_delete).

Endpoints

Endpoints are managed by your customers through the Portal, but you can also manage them programmatically.

Create Endpoint

from hookbase.models import CreateEndpointParams
 
# The signing secret is returned only here; give it to your customer.
endpoint = client.outbound.endpoints.create("519c05c7-ada5-5d3e-9b80-2c0dc9743117", CreateEndpointParams(
    url="https://example.com/webhooks",
    description="Production",
))
secret = endpoint.secret
 
# An Ed25519 endpoint: Hookbase keeps the private key and returns only the public key
ed_endpoint = client.outbound.endpoints.create("519c05c7-ada5-5d3e-9b80-2c0dc9743117", CreateEndpointParams(
    url="https://example.com/webhooks",
    signature_scheme="ed25519",
))
public_key = ed_endpoint.public_key  # whpk_...

List Endpoints

page = client.outbound.endpoints.list(application_id="519c05c7-ada5-5d3e-9b80-2c0dc9743117")

Endpoint Statistics

The Python SDK has no stats() call. Delivery counters are fields on the endpoint itself:

endpoint = client.outbound.endpoints.get("e25aaf14-ecb3-5f7c-9597-225979d06ed9")
print(endpoint.total_messages, endpoint.total_successes, endpoint.total_failures)
print(endpoint.avg_response_time_ms, endpoint.circuit_state)

Rotate the Signing Secret

# The old secret keeps signing for the grace period (seconds, 0-86400)
rotated = client.outbound.endpoints.rotate_secret("e25aaf14-ecb3-5f7c-9597-225979d06ed9", grace_period=3600)
print(rotated.secret, rotated.previous_secret_expires_at)

For an Ed25519 endpoint, rotated.secret is empty: read rotated.public_key, and rotated.previous_public_key for the key that keeps signing during the grace period.

Portal Tokens

Portal tokens authenticate the embedded Portal component.

Create Portal Token

from hookbase.models import CreatePortalTokenParams
 
token = client.outbound.portal_tokens.create("519c05c7-ada5-5d3e-9b80-2c0dc9743117", CreatePortalTokenParams(
    name="Customer dashboard",
    scopes=["read", "write"],
    expires_in_days=30,
))
 
# Return token.token to your frontend; it is only returned on create
plaintext = token.token

Webhook Verification

Help your customers verify incoming webhooks. Always verify the raw body: JSON that was parsed and re-serialized no longer matches the signature.

Verify a Request

import os
 
from flask import Flask, request
 
from hookbase import Webhook, WebhookVerificationError
 
app = Flask(__name__)
webhook = Webhook(os.environ["HOOKBASE_WEBHOOK_SECRET"])  # whsec_...
 
 
@app.post("/webhooks")
def receive() -> tuple[str, int]:
    try:
        event = webhook.verify(request.get_data(), dict(request.headers))
    except WebhookVerificationError:
        return "Invalid signature", 401
    print("Event:", event["type"])
    return "OK", 200

verify takes the headers as a dict or any mapping, so a framework's header object can be passed as it is. It checks the webhook-id, webhook-timestamp and webhook-signature headers (names are case-insensitive), rejects timestamps more than 5 minutes from now, and accepts any of the space-separated signatures sent during a secret rotation. It returns the parsed JSON body. Pass tolerance= (seconds) to change the time window.

Webhook verifies v1 (HMAC-SHA256) signatures with a whsec_ secret. A secret that is not valid base64 raises ValueError when the Webhook is created, not on the first request. For endpoints using the ed25519 scheme, pass the endpoint's whpk_ public key instead (Webhook(endpoint.public_key)), and verify checks v1a signatures. This needs the hookbase[ed25519] extra. If the key and the header don't match (a whsec_ secret against a header that only has v1a signatures, or the reverse), the WebhookVerificationError message says which key to use. See Ed25519 (v1a) Signatures.

Generate Test Headers

from hookbase import Webhook
 
webhook = Webhook("whsec_dGVzdF9zZWNyZXRfa2V5X2Zvcl90ZXN0aW5n")
 
payload = '{"orderId":"test_123"}'
headers = webhook.generate_test_headers(payload)
# Use these headers to test your webhook handler

A whpk_ public key can't sign. To produce v1a test headers, create a Webhook from an Ed25519 secret key in Standard Webhooks form, whsk_ followed by the base64 of the 32-byte seed and the 32-byte public key (one you generate yourself; Hookbase never returns the private key):

import base64
 
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
from hookbase import Webhook
 
key = Ed25519PrivateKey.generate()
seed = key.private_bytes(
    serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption()
)
public = key.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
 
signer = Webhook("whsk_" + base64.b64encode(seed + public).decode())
verifier = Webhook("whpk_" + base64.b64encode(public).decode())
 
payload = '{"orderId":"test_123"}'
event = verifier.verify(payload, signer.generate_test_headers(payload))

Error Handling

Every error the SDK raises is a HookbaseError. All of them can be imported from hookbase:

Exception	When
`APIError`	Any error response: `status_code`, `code`, `details`, `request_id`
`AuthenticationError`	401
`ForbiddenError`	403, including plan limits (`code` `LIMIT_EXCEEDED`) and disabled features (`FEATURE_DISABLED`)
`NotFoundError`	404
`ConflictError`	409
`ValidationError`	400 / 422, with `validation_errors` (field name to messages)
`RateLimitError`	429, with `retry_after` (seconds)
`TimeoutError`	The request timed out
`NetworkError`	Could not connect, the connection dropped, or `base_url` is not an absolute http(s) URL
`WebhookVerificationError`	Signature verification failed

On an APIError, status_code is the status the API returned (a ValidationError is 400 or 422), code is the code from the response body (or the class default, such as not_found, when the body has none), and details is the body's structured detail: {"fieldErrors": ..., "formErrors": ...} for a validation failure, or extra fields such as current and max on LIMIT_EXCEEDED. It is None when there is none. request_id is the x-request-id response header.

The status-specific errors subclass APIError. TimeoutError and NetworkError don't: they subclass HookbaseError directly. hookbase.TimeoutError is not the built-in TimeoutError, so import it from hookbase.

from hookbase import (
    APIError,
    ConflictError,
    NotFoundError,
    RateLimitError,
    ValidationError,
)
 
try:
    client.outbound.endpoints.get("invalid_id")
except NotFoundError:
    print("Endpoint not found")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except ValidationError as e:
    print(f"Invalid: {e.validation_errors}")
except ConflictError as e:
    print(f"Conflict: {e}")
except APIError as e:
    print(f"API error {e.status_code} {e.code}: {e} (request {e.request_id})")

Retries

Failed requests are retried up to max_retries times (default 3):

429 is retried for every method, after the Retry-After the API sends (seconds or an HTTP date), as long as that wait is at most 60 seconds. A 429 without a usable Retry-After waits 60 seconds before each retry. A longer Retry-After is not waited for: the RateLimitError is raised at once with retry_after set.
Any 5xx status (500 to 599), read timeouts and dropped connections are retried with exponential backoff (1, 2, 4 ... seconds, at most 10, plus up to 1 second of jitter) for every method except POST. A POST is retried only when it carries an idempotency key. Of the outbound calls, that is messages.send, which always has one.
A connection that was never made (refused, DNS failure, connect timeout) is always retried.
Other 4xx responses, including 409, are never retried.

Request Options

There are no per-call request options. Timeout and retries are set on the client, so create a second client for calls that need different values. The idempotency key is an argument of messages.send itself. To cancel an AsyncHookbase call, cancel its task or wrap it in asyncio.wait_for.

Pagination

list methods return a page. page.data holds this page's items (iterating the page yields the same items), page.has_more says whether another page follows, and page.next_page() fetches it. The outbound lists are cursor pages, which also have page.next_cursor:

page = client.outbound.endpoints.list(application_id="519c05c7-ada5-5d3e-9b80-2c0dc9743117", limit=100)
while True:
    for endpoint in page.data:
        print(endpoint.url)
    if not page.has_more:
        break
    page = page.next_page()
 
# The same, without the loop
for endpoint in client.outbound.endpoints.list(application_id="519c05c7-ada5-5d3e-9b80-2c0dc9743117").auto_paging_iter():
    print(endpoint.url)

With AsyncHookbase, next_page() is awaited and auto_paging_iter() is an async iterator:

page = await client.outbound.applications.list()
async for app in page.auto_paging_iter():
    print(app.name)

The inbound lists (sources, destinations, routes, transforms, filters) are numbered pages: they take page and page_size (at most 100) and have page.total and page.page instead of a cursor. schemas.list and cron_jobs.list return everything as one page. deliveries.list takes limit and offset. The same has_more, next_page() and auto_paging_iter() work on all of them.

sources.list still accepts search, provider and is_active, destinations.list search and is_active, and routes.list source_id, destination_id and is_active, but the API does not filter on them. Passing one raises a DeprecationWarning, and it is not sent.

portal_tokens.list, message_log.list_attempts, tunnels.list and organizations.list return a plain list.

Models

Responses are Pydantic v2 models in hookbase.models, named after the API's resources (Application, WebhookEndpoint, EventType, Subscription, OutboundMessage, OutboundAttempt, PortalToken, Source, Event, Delivery ...). Attributes are snake_case, and model_dump(by_alias=True) gives the API's camelCase JSON back. Timestamps are ISO-8601 strings, and free-form JSON such as payloads and metadata is a dict. Fields the API adds later are ignored rather than failing to decode. Outbound replay results are OutboundReplayResult in hookbase.models (ReplayResult there is the inbound one).

Inbound and Admin Resources

The SDK also covers the inbound API (sources, destinations, routes, events, deliveries, transforms, filters, schemas, provider catalog) and admin resources (analytics, API keys, tunnels, cron jobs, organizations), following the same pattern:

# Dashboard analytics: range is 1h, 24h, 7d (the default) or 30d
stats = client.analytics.dashboard(range="7d", tz="Europe/Berlin")
print(stats.overview.total_events, stats.overview.failed_deliveries)
print(stats.latency.percentiles.p95, [d.name for d in stats.destinations])
 
# Check a payload against a schema without sending it
result = client.schemas.validate("sch_123", {"orderId": "ord_123"})
for error in result.errors:
    print(error.path, error.message)
 
# A tunnel's auth token is returned once, on create
from hookbase.models import CreateTunnelParams
 
tunnel = client.tunnels.create(CreateTunnelParams(name="dev"))
print(tunnel.tunnel_url, tunnel.auth_token)
 
# Recent runs of a cron job
for run in client.cron_jobs.list_executions("crn_123", limit=10):
    print(run.status, run.response_status, run.error_message)

Filters take logic="AND" or "OR"; the SDK sends it uppercase whatever case you pass. transforms.update and organizations.update return the stored object, read back with a second request, because the API answers those updates with {"success": true} only.

The organizations resource and api_keys.create work only with a user session (a dashboard login), not with an API key: called with a whr_ key they raise ForbiddenError (403).

See the README and examples for more.