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

The same build is attached to each GitHub release as a wheel, for installs from a file: pip install ./hookchat-<version>-py3-none-any.whl.

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 read, and every key and webhook method, takes an optional tenant= keyword for operator keys. A bound key (the usual case) needs none. A send takes none either way: its conversation id says which workspace it belongs to.

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.

participant_handle, participant_username and participant_picture hold the other party's name, Instagram username and picture URL, each None when Meta does not share it. The lookup runs just after the first message arrives, theirs or one your team sent from the app, so participant_profile says where it stands: pending (not run yet), complete (something was stored and a participant.updated event went out) or unavailable (Meta shared nothing; it is tried again on their next message). Meta's picture URLs expire, so download the image if you need it later.

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)  # safe to repeat, returns True

A test-scope key cannot create 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)

The names an endpoint can subscribe to are in EVENT_TYPES; a name outside that list is refused with invalid_request before anything is stored.

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 your receiver will then see the same event id twice.

Realtime

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

The ticket is single use and bound to the workspace. 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, ParticipantEvent

# 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)
    elif isinstance(event, MessageEvent) and event.message.direction == "echo":
        # Sent from the Instagram or Messenger app. A send through the API is "outbound".
        handle_app_send(event.message.conversation_id, event.message.text)
    elif isinstance(event, ParticipantEvent):
        # Once per conversation, when the name, username or picture is first stored.
        p = event.participant
        save_profile(p.conversation_id, p.name, p.username, p.picture)
    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}

message.sent covers everything the account sends: event.message.direction is outbound for a send through the API and echo for one sent from the Instagram or Messenger app, which arrives a few seconds after it is sent. An endpoint subscribed to message.sent gets both; there is nothing to set up.

Events are MessageEvent (message.received, message.sent), MessageFailedEvent, TestEvent, AccountEvent (account.connected, account.disconnected) and ParticipantEvent (participant.updated). Deliveries are at least once and unordered; dedupe on event.id. message.received carries only the participant's id; participant.updated follows once per conversation with their name, username and picture, and can arrive before or after that message. 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 write, 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