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:
            hold = client.messages.send_as_human_agent(
                conversation.id, "Following up on your question.", actor_id="agent_42"
            )
            # Nothing is sent yet: agent_42 opens hold.verify_url and passes a check.

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.

Held human-agent sends

send_as_human_agent does not send at once. It returns a Hold with status awaiting_verification and a verify_url. The person who typed the message opens that page and passes a check in the browser, and then HookChat sends it. The check raises the cost of automation; it is not proof of a person. The hold waits up to 15 minutes, or less if the 7 day window ends sooner.

verify_url carries a one-time secret. Hand it only to the person who typed the message, and never log it. It comes only in the answer to the send; a read of the hold never has it.

hold = client.messages.send_as_human_agent("CONV#...", "Following up", actor_id="agent_42")
show_to_agent_42(hold.verify_url)  # your own UI; never log it

hold = client.messages.get_hold(hold.id)  # where it is now
hold = client.messages.wait_for_hold(hold.id)  # until it is sent
hold.platform_message_id

hold.status is awaiting_verification, sending, sent, failed or expired. The last three are final. wait_for_hold reads the hold every second, then every two seconds, and returns it once sent. A failed hold raises the error its error.code names (HumanAgentUnavailableError, RateLimitError, SendFailedError, NotFoundError or ServerError). An expired hold raises ConflictError with hold_expired: nobody passed the check, and nothing was sent. Pass timeout (seconds, default 15 minutes) to change the wait. Past it, wait_for_hold raises HoldTimeoutError with hold_id and last_status; the hold goes on, and get_hold reads it until it expires. An unknown or ended hold id raises NotFoundError with hold_not_found. AsyncHookChat has the same methods.

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. Nor do get_hold and wait_for_hold: the hold belongs to the key that made it.

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#...",
    attachments=[Attachment("image", "https://cdn.example.com/photo.jpg")],
    reply_to="mid.abc123",
)
client.messages.reply("CONV#...", "Here is the photo")

hold = client.messages.send_as_human_agent("CONV#...", "Following up", actor_id="agent_42")
hold.verify_url  # held until the agent passes the check; see above

Pass text or attachments, not both: a Meta message is text or media, so words and a picture take two calls. Up to 10 images go as one message; a video, audio or file goes alone. Any other shape raises ValueError before any request. Reads exclude test-scope rows unless you pass include_test=True.

Attachments

Upload a file from your computer, then send it by id:

from hookchat import Attachment

a = client.attachments.upload("CONV#...", "./photo.jpg")  # three steps, one call
client.messages.reply("CONV#...", attachments=[Attachment(type=a.type, attachment_id=a.id)])

file is a path, bytes, or a binary file object. The name and content type come from it; pass filename, content_type or type to override them. upload asks for an upload url (POST /v1/attachments/uploads), posts the file to it with no API key, then finishes (POST /v1/attachments), and returns the UploadedAttachment Meta now has. Meta's limits apply: 8 MB for an image, 25 MB for a video, audio or file. Too big raises ValidationError with attachment_too_large (413), and a type the account does not take unsupported_media_type (415). HookChat keeps the file only until Meta has it, at most a day. AsyncHookChat has the same await client.attachments.upload(...).

HookChat hands the file to Meta in the background, so upload waits: it reads the upload every second, then every two seconds, until Meta has it. Pass wait_timeout (seconds, default 120) to change the wait. Past it, upload raises UploadTimeoutError; the upload goes on, and client.attachments.get(exc.upload_id) reads it later. A file Meta refuses raises SendFailedError.

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