hookchat

The official Python SDK for HookChat, the Instagram and Messenger DM webhook gateway. One dependency (httpx), fully typed, sync and async clients with the same surface, automatic retries, and webhook signature verification.

Install

pip install hookchat
# or
uv add hookchat

Python 3.10 or newer.

Quickstart

from hookchat import HookChat, WindowClosedError

client = HookChat(api_key="hookchat_live_...")

for conversation in client.conversations.iterate(max_items=20):
    if not conversation.unanswered:
        continue
    try:
        client.messages.reply(conversation.id, "Thanks for reaching out, we are on it.")
    except WindowClosedError:
        if conversation.can_send_as_human_agent:
            client.messages.send_as_human_agent(
                conversation.id, "Following up on your question.", actor_id="agent_42"
            )

The client reads nothing from the environment. Pass the key explicitly, or wire it yourself:

import os

client = HookChat(api_key=os.environ["HOOKCHAT_API_KEY"])

Client options

HookChat(
    api_key="hookchat_live_...",
    base_url="https://hookchat.dev",  # the API origin
    timeout=30.0,  # seconds, or an httpx.Timeout
    max_retries=2,  # retry budget for retryable failures
    tenant=None,  # default ?tenant for operator keys
    default_headers=None,  # extra headers on every request
    http_client=None,  # reuse your own httpx.Client
)

Use the client as a context manager (with HookChat(...) as client:) or call client.close() to release the connection pool.

The send policy

There are two send paths and no generic send:

The gateway computes the window and exposes it on every conversation as window.state (open_24h, human_agent_only, closed), can_reply and can_send_as_human_agent. Never recompute it. Outside the 24 hour window reply raises WindowClosedError; past 7 days send_as_human_agent raises HumanAgentUnavailableError.

Resources

Every method takes an optional tenant= keyword for operator keys. A bound key (the usual case) needs none.

Ping

ping = client.ping()  # unauthenticated liveness probe
ping.name, ping.version, ping.time

Conversations

page = client.conversations.list(limit=50)  # Page[Conversation]
page.items, page.cursor
more = client.conversations.list(cursor=page.cursor)

for conversation in client.conversations.iterate(max_items=500):
    print(conversation.id, conversation.window.state, conversation.last_message_text)

detail = client.conversations.get("CONV#instagram#123#456")
detail.conversation.can_reply
for message in detail.messages:
    print(message.direction, message.text, message.sent_at)

Conversation ids contain #; the SDK percent-encodes them for you.

Messages

page = client.messages.list(conversation_id="CONV#...", direction="inbound", limit=20)
for message in client.messages.iterate(direction="outbound", max_items=100):
    ...
message = client.messages.get("mid.abc123")

sent = client.messages.reply("CONV#...", "Hello")
sent.platform_message_id

# Media, and threaded replies (a no-op where the platform does not support them).
from hookchat import Attachment

client.messages.reply(
    "CONV#...",
    "Here is the photo",
    attachments=[Attachment("image", "https://cdn.example.com/photo.jpg")],
    reply_to="mid.abc123",
)

client.messages.send_as_human_agent("CONV#...", "Following up", actor_id="agent_42")

text may be omitted when attachments is given. Reads exclude test-scope rows unless you pass include_test=True.

Accounts

for account in client.accounts.list():
    print(account.platform, account.handle, account.status, account.refresh_error)

Audit

page = client.audit.list(limit=100)
for entry in client.audit.iterate(max_items=1000):
    print(entry.at, entry.actor, entry.action, entry.resource)

Stats

stats = client.stats.get()
stats.messages_24h, stats.dlq_total, stats.window_hours
for row in stats.endpoints:
    print(row.endpoint_id, row.delivered, row.failed, row.dlq)

Keys

created = client.keys.create(label="ci", scope="test")  # scope defaults to the caller's
created.secret  # shown exactly once
for key in client.keys.list():
    print(key.id, key.prefix, key.scope, key.status)
client.keys.revoke(created.id)  # idempotent, returns True

A test-scope key cannot mint a live-scope key (PermissionDeniedError).

Webhooks

result = client.webhooks.create("https://example.com/hooks/hookchat", events=["message.received"])
endpoint = result.endpoint
secret = result.signing_secret  # shown exactly once, store it

client.webhooks.list()
client.webhooks.get(endpoint.id)
client.webhooks.update(endpoint.id, events=["message.received", "message.sent"])
client.webhooks.update(endpoint.id, status="paused")  # and "active" to resume
rotated = client.webhooks.rotate_secret(endpoint.id)  # old secret keeps verifying for 24 hours
client.webhooks.test(endpoint.id)  # sends a test.event through the real path
client.webhooks.delete(endpoint.id)

Omitting events subscribes the endpoint to every event type. Deleting tombstones the endpoint: the id and its delivery history are kept, the secrets are erased.

Deliveries

deliveries = client.webhooks.deliveries

page = deliveries.list(endpoint.id, limit=50)
for delivery in deliveries.iterate(endpoint.id, max_items=200):
    print(delivery.delivery_id, delivery.status, delivery.attempt_number, delivery.created_at)

report = deliveries.summary(endpoint.id)  # failing deliveries, last 24 hours
report = deliveries.summary(
    endpoint.id, since=datetime.now(timezone.utc) - timedelta(hours=6), status=["dlq", "failed"]
)
report.deliveries, report.truncated

deliveries.replay(endpoint.id, delivery_id)  # fresh attempt budget
deliveries.replay(endpoint.id, delivery_id, force=True)  # re-send one that already succeeded

result = deliveries.replay_all(endpoint.id, from_=datetime(2026, 9, 1, tzinfo=timezone.utc))
result.matched, result.enqueued, result.skipped
while result.cursor:
    result = deliveries.replay_all(endpoint.id, from_=..., cursor=result.cursor)

Delivery timestamps are epoch milliseconds, as the API returns them. Time arguments (since, from_, to) accept an aware datetime, an ISO-8601 string, or epoch milliseconds. replay_all defaults to the dlq and failed statuses; including delivered requires force=True, and the consumer will then see a duplicate event id.

Realtime

ticket = client.realtime.ticket()
ticket.ticket, ticket.expires_at, ticket.expires_in_ms

The ticket is single use and tenant bound. Open the WebSocket before it expires.

Anything else

data = client.request("GET", "/v1/some/new/path", query={"limit": 5})

Returns the envelope's data unparsed. GETs are retried like every other read.

Errors

Every API failure raises HookChatError or a subclass. Branch on the class, or on error.code, never on the message.

from hookchat import HookChatError, NotFoundError, WindowClosedError

try:
    client.messages.reply(conversation_id, text)
except WindowClosedError:
    ...
except NotFoundError:
    ...
except HookChatError as error:
    print(error.code, error.status, error.message, error.detail, error.request_id)

error.request_id carries the gateway's correlation id; quote it when reporting a problem. error.body holds the decoded response.

Webhook verification

Every delivery carries HookChat-Signature: t=<unix seconds>,v1=<hex>, where v1 is HMAC-SHA256 of "<t>.<raw body>" under your endpoint's signing secret. During a secret rotation the header carries two v1= digests and either secret verifies. verify_webhook checks the signature, rejects timestamps more than 300 seconds from now, and returns the typed event.

Always pass the raw request bytes. Re-serialising the JSON changes the bytes and breaks the signature.

from hookchat import verify_webhook, SignatureVerificationError, MessageEvent

# Flask
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = "whs_..."


@app.post("/hooks/hookchat")
def hookchat_webhook():
    try:
        event = verify_webhook(request.get_data(), request.headers, SECRET)
    except SignatureVerificationError:
        abort(400)
    if isinstance(event, MessageEvent) and event.type == "message.received":
        handle_inbound(event.message.conversation_id, event.message.text)
    return "", 200

# FastAPI
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()


@app.post("/hooks/hookchat")
async def hookchat_webhook(request: Request):
    body = await request.body()
    try:
        event = verify_webhook(body, request.headers, SECRET)
    except SignatureVerificationError:
        raise HTTPException(status_code=400)
    return {"received": event.id}

Events are MessageEvent (message.received, message.sent), MessageFailedEvent, TestEvent and AccountEvent (account.connected, account.disconnected). Deliveries are at least once and unordered; dedupe on event.id. An event type this SDK does not know parses as a plain Event with its raw data.

compute_signature(payload, timestamp, secret) is exported for building test fixtures.

Pagination

List methods return a Page with items and cursor. Pass the cursor back to fetch the next page, or use iterate, which follows the cursor and accepts max_items:

page = client.audit.list(limit=100)
while page.has_more:
    page = client.audit.list(limit=100, cursor=page.cursor)

for entry in client.audit.iterate(limit=100, max_items=5000):
    ...

iterate exists on conversations, messages, audit and webhooks.deliveries. Limits are clamped to 100 by the gateway.

Retries and timeouts

Reads (every GET) are retried on HTTP 429, 5xx, connection errors and timeouts with jittered exponential backoff (0.5 s base, 8 s cap), honouring Retry-After when present (capped at 60 s). webhooks.test, deliveries.replay and deliveries.replay_all are retried on 429 only. Every other mutation, including both send methods, is never retried automatically. The default budget is max_retries=2, so a read makes at most three attempts.

The default timeout is 30 seconds (10 to connect). Change either per call site without a new connection pool:

fast = client.with_options(timeout=5, max_retries=0)
fast.stats.get()

Async usage

AsyncHookChat has the same surface. Every method is a coroutine and iterate returns an async iterator.

import asyncio
from hookchat import AsyncHookChat


async def main():
    async with AsyncHookChat(api_key="hookchat_live_...") as client:
        stats = await client.stats.get()
        async for conversation in client.conversations.iterate(max_items=50):
            print(conversation.id)
        await asyncio.gather(
            client.messages.reply("CONV#a", "hi"),
            client.messages.reply("CONV#b", "hi"),
        )


asyncio.run(main())

Versioning

The SDK follows semantic versioning. Minor releases add fields, methods and event types; the models ignore unknown fields, so a newer gateway never breaks an older SDK. Breaking changes to the public API bump the major version. See CHANGELOG.md.

Development

uv sync
uv run ruff check .
uv run ruff format --check .
uv run mypy --strict src
uv run pytest -q                 # unit tests, no network
uv run pytest tests/e2e -q       # against a live gateway, see tests/e2e/conftest.py

License

MIT