posthook

The official Python client library for the Posthook API -- schedule webhooks and deliver them reliably.

Installation

pip install posthook-python

Requirements: Python 3.9+. Dependencies: httpx and websockets.

Quick Start

importposthookclient=posthook.Posthook("pk_...")
# Schedule a webhook 5 minutes from nowhook=client.hooks.schedule(
path="/webhooks/user-created",
post_in="5m",
data={"userId": "123", "event": "user.created"},
)
print(hook.id) # UUIDprint(hook.status) # "pending"

How It Works

Your Posthook project has a domain configured in the dashboard (e.g., webhook.example.com). When you schedule a hook, you specify a path (e.g., /webhooks/user-created). At the scheduled time, Posthook delivers the hook by POSTing to the full URL (https://webhook.example.com/webhooks/user-created) with your data payload and signature headers.

Authentication

You can find your API key under Project Settings in the Posthook dashboard. Pass it directly to the constructor, or set the POSTHOOK_API_KEY environment variable:

# Explicit API keyclient=posthook.Posthook("pk_...")
# From environment variableclient=posthook.Posthook() # reads POSTHOOK_API_KEY

For webhook signature verification, also provide a signing key:

client=posthook.Posthook("pk_...", signing_key="ph_sk_...")

The signing key can also be set via the POSTHOOK_SIGNING_KEY environment variable.

Scheduling Hooks

Three mutually exclusive scheduling modes are available. You must provide exactly one of post_in, post_at, or post_at_local.

Relative delay (post_in)

Schedule after a relative delay. Accepts s (seconds), m (minutes), h (hours), or d (days):

hook=client.hooks.schedule(
path="/webhooks/send-reminder",
post_in="30m",
data={"userId": "123"},
)

Absolute UTC time (post_at)

Schedule at an exact UTC time. Accepts datetime objects or RFC 3339 strings:

fromdatetimeimportdatetime, timedelta, timezone# Using a datetime object (automatically converted to UTC)hook=client.hooks.schedule(
path="/webhooks/send-reminder",
post_at=datetime.now(timezone.utc) +timedelta(hours=1),
data={"userId": "123"},
)
# Using an RFC 3339 stringhook=client.hooks.schedule(
path="/webhooks/send-reminder",
post_at="2026-06-15T10:00:00Z",
data={"userId": "123"},
)

Local time with timezone (post_at_local)

Schedule at a local time. Posthook handles DST transitions automatically:

hook=client.hooks.schedule(
path="/webhooks/daily-digest",
post_at_local="2026-03-01T09:00:00",
timezone="America/New_York",
data={"userId": "123"},
)

Custom retry configuration

Override your project's default retry behavior for a specific hook:

hook=client.hooks.schedule(
path="/webhooks/critical",
post_in="1m",
data={"orderId": "456"},
retry_override=posthook.HookRetryOverride(
min_retries=10,
delay_secs=15,
strategy="exponential",
backoff_factor=2.0,
max_delay_secs=3600,
jitter=True,
),
)

Managing Hooks

Get a hook

hook=client.hooks.get("hook-uuid")

List hooks

hooks=client.hooks.list(status=posthook.STATUS_FAILED, limit=50)
print(f"Found {len(hooks)} hooks")

All list parameters are optional:

Parameter	Description
status	Filter by status: "pending", "retry", "completed", "failed"
limit	Max results per page
sort_by	Sort field (e.g., "createdAt", "postAt")
sort_order	"ASC" or "DESC"
post_at_before	Filter hooks scheduled before this time (ISO string)
post_at_after	Cursor: hooks scheduled after this time (ISO string)
created_at_before	Filter hooks created before this time (ISO string)
created_at_after	Filter hooks created after this time (ISO string)

Cursor-based pagination

Use post_at_after as a cursor. After each page, advance it to the last hook's post_at:

limit=100cursor=NonewhileTrue:
hooks=client.hooks.list(status="failed", limit=limit, post_at_after=cursor)
forhookinhooks:
print(hook.id, hook.failure_error)
iflen(hooks) <limit:
break# last pagecursor=hooks[-1].post_at.isoformat()

Auto-paginating iterator (list_all)

For convenience, list_all yields every matching hook across all pages automatically:

forhookinclient.hooks.list_all(status="failed"):
process(hook)

The async client returns an async iterator:

asyncforhookinclient.hooks.list_all(status="failed"):
awaitprocess(hook)

Delete a hook

To cancel a pending hook, delete it before delivery. Idempotent -- returns None on both 200 (deleted) and 404 (already deleted):

client.hooks.delete("hook-uuid")

Bulk Operations

Three bulk operations are available, each supporting by-IDs or by-filter:

• Retry -- Re-attempts delivery for failed hooks
• Replay -- Re-delivers completed hooks (useful for reprocessing)
• Cancel -- Cancels pending hooks before delivery

By IDs

result=client.hooks.bulk.retry(["id-1", "id-2", "id-3"])
print(f"Retried {result.affected} hooks")

By filter

result=client.hooks.bulk.cancel_by_filter(
start_time="2026-02-01T00:00:00Z",
end_time="2026-02-22T00:00:00Z",
limit=500,
endpoint_key="/webhooks/deprecated",
)
print(f"Cancelled {result.affected} hooks")

All six methods:

# By IDsclient.hooks.bulk.retry(hook_ids)
client.hooks.bulk.replay(hook_ids)
client.hooks.bulk.cancel(hook_ids)
# By filterclient.hooks.bulk.retry_by_filter(start_time, end_time, limit, ...)
client.hooks.bulk.replay_by_filter(start_time, end_time, limit, ...)
client.hooks.bulk.cancel_by_filter(start_time, end_time, limit, ...)

Filter methods also accept optional endpoint_key and sequence_id keyword arguments.

Verifying Webhook Signatures

When Posthook delivers a hook to your endpoint, it includes signature headers for verification. Use parse_delivery to verify and parse the delivery.

Important: You must pass the raw request body (bytes or string), not a parsed JSON object.

Flask

fromflaskimportFlask, requestimportposthookapp=Flask(__name__)
client=posthook.Posthook("pk_...", signing_key="ph_sk_...")
@app.route("/webhooks/user-created", methods=["POST"])defhandle_webhook():
try:
delivery=client.signatures.parse_delivery(
body=request.get_data(),
headers=dict(request.headers),
)
exceptposthook.SignatureVerificationError:
return"invalid signature", 401print(delivery.hook_id) # from Posthook-Id headerprint(delivery.path) # "/webhooks/user-created"print(delivery.data) # your custom data payloadprint(delivery.post_at) # when it was scheduledprint(delivery.posted_at) # when it was deliveredreturn"", 200

Django

fromdjango.httpimportHttpResponseimportposthookclient=posthook.Posthook("pk_...", signing_key="ph_sk_...")
defhandle_webhook(request):
try:
delivery=client.signatures.parse_delivery(
body=request.body,
headers=dict(request.headers),
)
exceptposthook.SignatureVerificationError:
returnHttpResponse(status=401)
print(delivery.hook_id)
print(delivery.data)
returnHttpResponse(status=200)

FastAPI

fromfastapiimportFastAPI, Request, Responseimportposthookapp=FastAPI()
client=posthook.Posthook("pk_...", signing_key="ph_sk_...")
@app.post("/webhooks/user-created")asyncdefhandle_webhook(request: Request):
body=awaitrequest.body()
try:
delivery=client.signatures.parse_delivery(
body=body,
headers=dict(request.headers),
)
exceptposthook.SignatureVerificationError:
returnResponse(status_code=401)
print(delivery.hook_id)
print(delivery.data)
returnResponse(status_code=200)

Custom tolerance

By default, signatures older than 5 minutes are rejected. You can override this:

delivery=client.signatures.parse_delivery(
body=raw_body,
headers=headers,
tolerance=600, # 10 minutes, in seconds
)

WebSocket Listener

The WebSocket listener receives hooks in real time without running an HTTP server. Use AsyncPosthook and call hooks.listen() with an async handler:

importasyncioimportposthookasyncdefmain():
asyncwithposthook.AsyncPosthook("pk_...") asclient:
asyncdefon_hook(delivery):
print(f"Received: {delivery.hook_id} -> {delivery.path}")
print(f"Data: {delivery.data}")
returnposthook.Result.ack()
listener=awaitclient.hooks.listen(
on_hook,
on_connected=lambdainfo: print(f"Connected: {info.project_name}"),
)
awaitlistener.wait() # Blocks until closedasyncio.run(main())

Result types

Your handler must return a Result:

Factory	Effect
Result.ack()	Processing complete — hook is marked as delivered immediately
Result.nack(error?)	Reject — triggers retry according to project settings
Result.accept(timeout)	Async — you have timeout seconds to call back via HTTP (see below)

returnposthook.Result.ack()
returnposthook.Result.nack("processing failed")
returnposthook.Result.accept(timeout=120)

Async processing with accept

Use accept when your handler needs more time than the 10-second ack window. After returning accept, POST to the callback URLs on the delivery to report the outcome:

asyncdefon_hook(delivery):
# Kick off background work, save the callback URLsawaitqueue.enqueue("process", {
"data": delivery.data,
"ack_url": delivery.ack_url,
"nack_url": delivery.nack_url,
})
returnposthook.Result.accept(timeout=300) # 5 minutes to call back# Later, in the background worker:awaitposthook.async_ack(job["ack_url"])
# or on failure:awaitposthook.async_nack(job["nack_url"], {"error": "failed"})

If neither URL is called before the deadline, the hook is retried.

Concurrency

By default, handlers run with unlimited concurrency (matching HTTP delivery behavior). Set max_concurrency to limit parallel handlers — deliveries that arrive while at capacity are nacked immediately so the server can retry them:

listener=awaitclient.hooks.listen(on_hook, max_concurrency=10)

Lifecycle callbacks

listener=awaitclient.hooks.listen(
on_hook,
on_connected=lambdainfo: print(f"Connected: {info.connection_id}"),
on_disconnected=lambdaerr: print(f"Disconnected: {err}"),
on_reconnecting=lambdaattempt: print(f"Reconnecting (attempt {attempt})"),
)

Stream API

For manual control over ack/nack, use hooks.stream() which returns an async iterator:

asyncwithposthook.AsyncPosthook("pk_...") asclient:
asyncwithawaitclient.hooks.stream() asstream:
asyncfordeliveryinstream:
print(delivery.hook_id, delivery.data)
ifshould_process(delivery):
awaitstream.ack(delivery.hook_id)
else:
awaitstream.nack(delivery.hook_id, "not ready")

HTTP fallback

If your project has a domain configured, hooks are delivered via HTTP when no WebSocket listener is connected. You can run both an HTTP endpoint and a WebSocket listener — the server uses WebSocket when available and falls back to HTTP automatically. Since both paths use the same Result type, you can share your handler logic:

asyncdefprocess_hook(delivery):
awaitprocess_order(delivery.data)
returnposthook.Result.ack()
# HTTP delivery (ASGI endpoint)app=signatures.asgi_handler(process_hook)
# WebSocket delivery (runs alongside)listener=awaitclient.hooks.listen(process_hook)

WebSocket delivery metadata

Deliveries received via WebSocket include a ws field with attempt info:

asyncdefon_hook(delivery):
ifdelivery.ws:
print(f"Attempt {delivery.ws.attempt}/{delivery.ws.max_attempts}")
ifdelivery.ws.forward_request:
print(f"Original body: {delivery.ws.forward_request.body}")
returnposthook.Result.ack()

ASGI/WSGI Handlers

For quick integration without a full web framework, SignaturesService provides handler wrappers:

ASGI

importposthooksignatures=posthook.create_signatures("ph_sk_...")
asyncdefon_hook(delivery):
print(delivery.data)
returnposthook.Result.ack()
# Mount as an ASGI endpoint (e.g. with uvicorn)app=signatures.asgi_handler(on_hook)

WSGI

importposthooksignatures=posthook.create_signatures("ph_sk_...")
defon_hook(delivery):
print(delivery.data)
returnposthook.Result.ack()
# Mount as a WSGI endpoint (e.g. with gunicorn)app=signatures.wsgi_handler(on_hook)

Async Hooks

When async hooks are enabled, parse_delivery() populates ack_url and nack_url on the delivery object. Return 202 from your handler and call back when processing completes.

FastAPI

fromfastapiimportFastAPI, Request, BackgroundTasks, HTTPExceptionfromfastapi.responsesimportResponseimportposthookapp=FastAPI()
client=posthook.Posthook("pk_...", signing_key="ph_sk_...")
asyncdefprocess_and_ack(delivery):
try:
awaitprocess_video(delivery.data["video_id"])
result=awaitposthook.async_ack(delivery.ack_url)
print(f"Applied: {result.applied}")
exceptExceptionase:
awaitposthook.async_nack(delivery.nack_url, {"error": str(e)})
@app.post("/webhooks/process-video")asyncdefhandle_webhook(request: Request, background_tasks: BackgroundTasks):
body=awaitrequest.body()
try:
delivery=client.signatures.parse_delivery(body=body, headers=dict(request.headers))
exceptposthook.SignatureVerificationError:
raiseHTTPException(status_code=401)
background_tasks.add_task(process_and_ack, delivery)
returnResponse(status_code=202)

Callback functions

The SDK provides standalone callback functions -- pass the URL from the delivery object:

# Sync (Flask, Django, background workers)result=posthook.ack(delivery.ack_url)
result=posthook.nack(delivery.nack_url, {"error": "processing failed"})
# Async (FastAPI, etc.)result=awaitposthook.async_ack(delivery.ack_url)
result=awaitposthook.async_nack(delivery.nack_url, {"error": "processing failed"})

Both return a CallbackResult:

result=posthook.ack(delivery.ack_url)
print(result.applied) # True if state changed, False if already resolvedprint(result.status) # "completed", "not_found", "conflict", etc.

ack() and nack() return normally for 200, 404, and 409 responses. They raise CallbackError for 401 (invalid token) and 410 (expired).

If processing happens in a separate worker, use the raw callback URLs instead:

queue.enqueue("transcode", {
"video_id": delivery.data["video_id"],
"ack_url": delivery.ack_url,
"nack_url": delivery.nack_url,
})

Error Handling

All API errors extend PosthookError and can be caught with isinstance or except:

importposthooktry:
hook=client.hooks.get("hook-id")
exceptposthook.RateLimitError:
print("Rate limited, retry later")
exceptposthook.AuthenticationError:
print("Invalid API key")
exceptposthook.NotFoundError:
print("Hook not found")
exceptposthook.PosthookErroraserr:
print(f"API error: {err.message} (status={err.status_code})")

Error class	HTTP Status	Code
BadRequestError	400	bad_request
AuthenticationError	401	authentication_error
ForbiddenError	403	forbidden
NotFoundError	404	not_found
PayloadTooLargeError	413	payload_too_large
RateLimitError	429	rate_limit_exceeded
InternalServerError	5xx	internal_error
PosthookConnectionError	--	connection_error
SignatureVerificationError	--	signature_verification_error

Configuration

client=posthook.Posthook(
"pk_...",
base_url="https://api.staging.posthook.io",
timeout=60,
signing_key="ph_sk_...",
)

Option	Description	Default
api_key	Your Posthook API key	POSTHOOK_API_KEY env var
base_url	Custom API base URL	https://api.posthook.io
timeout	Request timeout in seconds	30
signing_key	Signing key for webhook verification	POSTHOOK_SIGNING_KEY env var
http_client	Custom httpx.Client instance	--

Quota Info

After scheduling a hook, quota information is available on the returned Hook object:

hook=client.hooks.schedule(path="/test", post_in="5m")
ifhook.quota:
print(f"Limit: {hook.quota.limit}")
print(f"Usage: {hook.quota.usage}")
print(f"Remaining: {hook.quota.remaining}")
print(f"Resets at: {hook.quota.resets_at}")

Async Client

The AsyncPosthook client provides an identical API -- just await each call:

importposthookasyncwithposthook.AsyncPosthook("pk_...") asclient:
hook=awaitclient.hooks.schedule(path="/test", post_in="5m")
print(hook.id)
hooks=awaitclient.hooks.list(status="pending")

Both the sync and async clients support context managers for automatic cleanup:

# Syncwithposthook.Posthook("pk_...") asclient:
hook=client.hooks.schedule(path="/test", post_in="5m")
# Asyncasyncwithposthook.AsyncPosthook("pk_...") asclient:
hook=awaitclient.hooks.schedule(path="/test", post_in="5m")

You can also call close() / await close() manually if you prefer.

Debug Logging

The SDK logs all requests via Python's logging module under the "posthook" logger. Enable it to see request details:

importlogginglogging.basicConfig(level=logging.DEBUG)

Example output:

DEBUG:posthook:POST /v1/hooks -> 200 (0.153s)
DEBUG:posthook:GET /v1/hooks -> 200 (0.089s)

Advanced

Proxy support

Pass a custom httpx.Client configured with a proxy:

importhttpximportposthookhttp_client=httpx.Client(proxy="http://proxy.example.com:8080")
client=posthook.Posthook("pk_...", http_client=http_client)

Custom CA certificates

importhttpximportposthookhttp_client=httpx.Client(verify="/path/to/custom-ca-bundle.crt")
client=posthook.Posthook("pk_...", http_client=http_client)

Custom httpx client

For full control over HTTP behavior, provide your own httpx.Client (sync) or httpx.AsyncClient (async). The SDK will add its authentication headers automatically:

importhttpximportposthookhttp_client=httpx.Client(
timeout=60,
verify=True,
proxy="http://proxy.example.com:8080",
limits=httpx.Limits(max_connections=20),
)
client=posthook.Posthook("pk_...", http_client=http_client)

When you provide a custom client, the SDK does not close it on client.close() -- you are responsible for its lifecycle.

Resources

• Documentation — guides, concepts, and patterns
• API Reference — endpoint specs and examples
• Quickstart — get started in under 2 minutes
• Pricing — free tier included
• Status — uptime and incident history

Requirements

• Python 3.9+
• httpx >= 0.25.0
• websockets >= 12.0 (for WebSocket listener/stream)