Python SDK
hookchat is the official Python SDK. One dependency (httpx), fully typed, a sync client and an async client with the same surface, automatic retries on reads, and verify_webhook for the deliveries the gateway sends you.
Install#
pip install hookchatPython 3.10 or newer. uv add hookchat works too. The package is on the downloads page along with its changelog.
Create a client#
The client reads nothing from the environment; pass the key explicitly. Use it as a context manager (with HookChat(...) as client:) or call client.close() to release the connection pool.
import os
from hookchat import HookChat
client = HookChat(
api_key=os.environ["HOOKCHAT_API_KEY"], # hookchat_live_... or hookchat_test_...
base_url=os.environ["HOOKCHAT_BASE_URL"], # the API origin, https://hookchat.dev by default
timeout=30.0, # seconds, or an httpx.Timeout
max_retries=2, # retry budget for retryable failures
tenant=None, # default ?tenant for operator keys
)
ping = client.ping() # unauthenticated liveness probe
print("ok", ping.name, ping.version, ping.time)| Argument | Default | Meaning |
|---|---|---|
api_key | required | Sent as a bearer token on every call. |
base_url | https://hookchat.dev | The API origin. |
timeout | 30.0 | Seconds (10 to connect), or an httpx.Timeout. |
max_retries | 2 | Retry budget for retryable failures. 0 disables retries. |
tenant | None | Default ?tenant for operator keys. Bound keys need none. |
default_headers | None | Extra headers on every request. |
http_client | None | Reuse your own httpx.Client (proxies, custom transports). The SDK will not close it. |
The send policy#
Two send paths and no generic send. reply works inside 24 hours of the last inbound message; send_as_human_agent works from 24 hours to 7 days and takes actor_id as a required keyword. Outside the window reply raises WindowClosedError; past 7 days send_as_human_agent raises HumanAgentUnavailableError. text may be omitted when attachments is given, and reply_to threads under a platform message id.
import os
from hookchat import HookChat
client = HookChat(api_key=os.environ["HOOKCHAT_API_KEY"], base_url=os.environ["HOOKCHAT_BASE_URL"])
conversation_id = os.environ["CONVERSATION_ID"]
# Inside the 24 hour window.
reply = client.messages.reply(conversation_id, "Thanks, on it.")
print("reply sent", reply.platform_message_id)
# Between 24 hours and 7 days, typed by a human. actor_id is required.
follow_up = client.messages.send_as_human_agent(conversation_id, "Following up.", actor_id="agent-7")
print("human-agent sent", follow_up.platform_message_id)Resources#
Every method takes an optional tenant= keyword for operator keys. Conversation ids contain #; the SDK percent-encodes them for you. Reads exclude test-scope rows unless you pass include_test=True.
| Resource | Methods | Routes |
|---|---|---|
client.ping() | GET /v1/ping | |
client.conversations | list, iterate, get | GET /v1/conversations, GET /v1/conversations/{id} |
client.messages | reply, send_as_human_agent, list, iterate, get | POST /v1/messages/reply, POST /v1/messages/human-agent, GET /v1/messages, GET /v1/messages/{id} |
client.accounts | list | GET /v1/accounts |
client.audit | list, iterate | GET /v1/audit |
client.stats | get | GET /v1/stats |
client.keys | create, list, revoke | POST /v1/keys, GET /v1/keys, DELETE /v1/keys/{id} |
client.webhooks | create, list, get, update, delete, rotate_secret, test | /v1/webhooks, /v1/webhooks/{id}, .../rotate, .../test |
client.webhooks.deliveries | list, iterate, summary, replay, replay_all | /v1/webhooks/{id}/deliveries, .../summary, .../{delivery_id}/replay, .../replay |
client.realtime | ticket | POST /v1/realtime/ticket |
client.request(method, path) | Escape hatch for any /v1 path; returns the envelope's data unparsed. |
import os
from hookchat import HookChat
client = HookChat(api_key=os.environ["HOOKCHAT_API_KEY"], base_url=os.environ["HOOKCHAT_BASE_URL"])
accounts = client.accounts.list() # not paginated
for account in accounts:
if account.refresh_error:
print(account.handle, "needs relinking:", account.refresh_error)
stats = client.stats.get()
print("messages_24h", stats.messages_24h, "dlq_total", stats.dlq_total, "endpoints", len(stats.endpoints))
for entry in client.audit.list(limit=5): # a Page iterates over its items
print(entry.at, entry.actor, entry.action, entry.resource)
endpoints = client.webhooks.list() # never returns a secret
print(len(endpoints), "endpoints,", len(accounts), "accounts")Errors#
Every API failure raises HookChatError or a subclass carrying code, status, message, detail, request_id and body. Catch the family class or branch on code; never parse the message.
| Exception | When |
|---|---|
AuthenticationError | 401, the key is missing or invalid |
PermissionDeniedError | 403, the key is not scoped to that tenant |
ValidationError | 400, malformed request or tenant missing on an operator key |
NotFoundError | 404, no such conversation, message, endpoint, key or delivery |
ConflictError | 409, the resource's state refused the request |
WindowClosedError | 409 window_closed, subclass of ConflictError |
HumanAgentUnavailableError | 409 human_agent_unavailable |
MissingActorError | 409 missing_actor |
ReplayConflictError | 409 replay_conflict |
RateLimitError | 409 rate_limited (send budget) or an HTTP 429 |
SendFailedError | 502 send_failed, Meta or the network refused; detail has the reason |
ServerError | any other 5xx |
APIConnectionError | the gateway could not be reached after retries |
APITimeoutError | the request timed out after retries |
import os
from hookchat import HookChat, HookChatError, NotFoundError, WindowClosedError
client = HookChat(api_key=os.environ["HOOKCHAT_API_KEY"], base_url=os.environ["HOOKCHAT_BASE_URL"])
try:
client.messages.reply(os.environ["CLOSED_CONVERSATION_ID"], "Hello again")
except WindowClosedError as error:
print("refused:", error.code, "status", error.status, "request", error.request_id)
except NotFoundError as error:
print("no such conversation:", error.code)
except HookChatError as error:
print(error.code, error.status, error.message, error.detail)Webhook verification#
verify_webhook(payload, headers, secret, tolerance=300) checks the HookChat-Signature digest over the raw bytes, rejects timestamps more than 300 seconds from now, accepts either digest during a rotation overlap, and returns the typed event. It raises SignatureVerificationError, which is deliberately not a HookChatError because it carries no HTTP status. Always pass the raw request bytes. compute_signature(payload, timestamp, secret) is exported for building fixtures.
import os
from http.server import BaseHTTPRequestHandler, HTTPServer
from hookchat import MessageEvent, SignatureVerificationError, verify_webhook
SECRET = os.environ["WEBHOOK_SECRET"]
class Receiver(BaseHTTPRequestHandler):
def do_POST(self):
raw = self.rfile.read(int(self.headers.get("Content-Length", "0"))) # the exact bytes
try:
event = verify_webhook(raw, dict(self.headers.items()), SECRET)
except SignatureVerificationError:
self.send_response(401)
self.end_headers()
return
if isinstance(event, MessageEvent) and event.type == "message.received":
print("inbound", event.message.conversation_id, event.message.text, flush=True)
else:
print("verified", event.type, event.id, flush=True)
self.send_response(200)
self.end_headers()
HTTPServer(("", int(os.environ["PORT"])), Receiver).serve_forever()Events are MessageEvent (message.received, message.sent), MessageFailedEvent, TestEvent and AccountEvent. A type this SDK does not know yet parses as a plain Event with its raw data, so a newer gateway never breaks an older consumer. Deliveries are at least once and unordered; dedupe on event.id. In Flask use request.get_data(), in FastAPI await request.body().
Pagination#
List methods return a Page with items, cursor and has_more. Pass the cursor back, or use iterate, which follows it and accepts max_items. iterate exists on conversations, messages, audit and webhooks.deliveries. Limits are clamped to 100 by the gateway.
import os
from hookchat import HookChat
client = HookChat(api_key=os.environ["HOOKCHAT_API_KEY"], base_url=os.environ["HOOKCHAT_BASE_URL"])
unanswered = 0
for conversation in client.conversations.iterate(limit=100, max_items=500):
if conversation.unanswered and conversation.can_reply:
unanswered += 1
print(unanswered, "unanswered conversations still inside the window")
page = client.audit.list(limit=3)
print(len(page), "audit entries, has_more:", page.has_more)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. with_options changes the timeout, the retry budget or the tenant per call site without a new connection pool.
import os
from hookchat import HookChat
client = HookChat(api_key=os.environ["HOOKCHAT_API_KEY"], base_url=os.environ["HOOKCHAT_BASE_URL"], max_retries=3)
fast = client.with_options(timeout=5, max_retries=0)
stats = fast.stats.get()
print("window_hours", stats.window_hours)Async usage#
AsyncHookChat has the same surface. Every method is a coroutine and iterate returns an async iterator.
import asyncio
import os
from hookchat import AsyncHookChat
async def main():
async with AsyncHookChat(
api_key=os.environ["HOOKCHAT_API_KEY"], base_url=os.environ["HOOKCHAT_BASE_URL"]
) as client:
stats, accounts = await asyncio.gather(client.stats.get(), client.accounts.list())
print("messages_24h", stats.messages_24h, "accounts", len(accounts))
async for conversation in client.conversations.iterate(max_items=5):
print(conversation.id, conversation.window.state)
print("async done")
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. hookchat.__version__ reports the installed version.