Skip to main content

Python SDK (hatang)

hatang is the server-side client for the Ha Tang API in Python. It covers what a backend does most: put plans and purchases on the customer's profile, sync the AI assistant's knowledge, answer conversations, send email, create VietQR payments, and verify webhooks and IPN calls. There is a sync client (Hatang) and an asyncio one (AsyncHatang) with the same methods.

Version

hatang 0.1.1 · Python 3.9+ · PyPI · changelog: CHANGELOG.md in the package

Before you start​

  • An API key (ht_...), created in the dashboard (see Authentication), with the scopes you call: crm:write for plans and purchases, chat:write for conversations and knowledge, email:send for email.
  • For the chat identity hash, the site's identity secret (Support → Channels → the site → Widget).

Install​

pip install hatang
# or
uv add hatang
# or
poetry add hatang

Create a client​

from hatang import Hatang

hatang = Hatang()  # reads HATANG_API_KEY (and HATANG_BASE_URL, HATANG_ENVIRONMENT)
hatang = Hatang("ht_...", base_url="http://localhost:3001", timeout=10)

Close it when you are done (hatang.close(), or with Hatang() as hatang:), or keep one for the life of the process. Pass your own httpx.Client as http_client for proxies, retries or tests.

asyncio (FastAPI, aiohttp, Starlette):

from hatang import AsyncHatang

hatang = AsyncHatang()

async def on_paid(order):
    await hatang.chat.purchases.upsert("web", order.id, {...})

Results and errors​

Every method returns the API's JSON as a dict. A call that does not succeed raises HatangError:

from hatang import HatangError

try:
    hatang.chat.purchases.upsert("web", order.id, body)
except HatangError as e:
    e.status_code  # HTTP status; 0 when Ha Tang was not reached (network, timeout)
    e.name         # validation_error, not_found, rate_limit_exceeded, ...
    e.retryable    # True for unreachable, 429 and 5xx: retry those later

This is the Python habit; the Node SDK returns { data, error } instead. Field names are the API's snake_case in both.

Plans and purchases​

Put what a customer pays for on their profile, so agents and the AI assistant can answer "when does my plan renew?" from facts. Use the same user id you pass to identify in the chat.

# A plan sold on your website (or one you verified with the App Store / Google Play yourself).
hatang.chat.subscriptions.upsert("web", subscription.id, {
    "status": "active",
    "sku": "PRO-1M",                      # the catalog product's SKU
    "external_user_id": str(user.id),
    "started_at": "2026-10-01T00:00:00Z",
    "current_period_end": "2026-11-01T00:00:00Z",
    "auto_renew": False,
    "amount": 99000, "currency": "VND",
})

# One-time purchases: credit packs, unlocks, courses. Send status="refunded" for the same id on a refund.
hatang.chat.purchases.upsert("web", order.id, {
    "store_product_id": "credits:300",    # your own code on the web; the store's product id in an app
    "sku": "XU-300",
    "kind": "consumable",
    "external_user_id": str(user.id),
    "amount": 29000, "currency": "VND",
    "purchased_at": "2026-10-30T10:00:00Z",
})
hatang.chat.purchases.list(external_user_id=str(user.id))

Store purchases use platform ios (Apple originalTransactionId for plans, transactionId for one-time purchases) or android (Play purchase token, or order id). Sending the same record again only updates it, so a backend can re-send everything it knows after each change. See In-app purchases and subscriptions.

Knowledge​

hatang.knowledge.sync("help", [
    {"external_id": f"help:{a.slug}", "title": a.title, "content": a.body, "url": a.url, "locale": "vi"}
    for a in articles
])  # the collection then holds exactly these; unchanged documents are not re-indexed
hatang.knowledge.search("đổi trả trong bao lâu", locale="vi")

See Knowledge API.

Conversations and customers​

convs = hatang.chat.conversations.list(status="open", unanswered=True)
hatang.chat.conversations.send(conv_id, {"content": "Đơn của bạn đã được gửi."}, idempotency_key=f"shipped:{order.id}")
hatang.chat.conversations.send(conv_id, {"content": "Khách VIP", "visibility": "private"})  # internal note
hatang.chat.customers.update(customer_id, {"stage": "won", "fields": {"plan": "pro"}})
csv_text = hatang.chat.customers.export_csv({"stage": "won"})

Products and orders: hatang.chat.products (list, search, get, create, update, remove) and hatang.chat.orders (list, get, create, update, payment_link).

Email​

from is a Python keyword, so write the body as a dict literal:

hatang.emails.send({
    "from": "Shop <[email protected]>",
    "to": "[email protected]",
    "subject": "Đơn hàng DH12",
    "html": "<p>Cảm ơn bạn!</p>",
    "attachments": [{"filename": "invoice.pdf", "content": pdf_bytes}],  # bytes are base64-encoded for you
}, idempotency_key=f"order-mail:{order.id}")
hatang.batch.send([...])  # up to 100

Payments and IPN​

payment = hatang.payments.create({"amount": 199000, "description": "DH12", "ipn_url": "https://shop.vn/hatang/ipn"})

from hatang import construct_ipn_event

def ipn(request):  # Django; pass the raw body before parsing it
    event = construct_ipn_event(request.body, request.headers, settings.HATANG_SECRET_KEY)
    if event["data"]["status"] == "paid":
        mark_paid(event["data"]["metadata"]["order_id"])

Webhooks​

from hatang import HatangSignatureError, construct_webhook_event

@app.post("/hatang/webhook")          # FastAPI
async def webhook(request: Request):
    try:
        event = construct_webhook_event(await request.body(), request.headers, WEBHOOK_SECRET)  # whsec_...
    except HatangSignatureError:
        raise HTTPException(400)
    if event["type"] == "chat.message.created":
        ...

Deliveries older than 5 minutes are refused (tolerance_sec changes it). verify_webhook and verify_ipn return a bool instead of raising.

Identity hash and purchase token​

from hatang import identity_hash, purchase_token

identity_hash(str(user.id), settings.HATANG_IDENTITY_SECRET)  # return {userId, hash} to the widget or the app
purchase_token("site_...", str(user.id))  # StoreKit appAccountToken / Play obfuscatedAccountId

Both match Ha Tang's servers and the app SDKs value for value. See Identity verification.

Types​

Request bodies are TypedDicts in hatang.types (ChatSubscriptionOptions, ChatPurchaseOptions, KnowledgeDocumentInput, CreateEmailOptions, CreatePaymentOptions…), so mypy and your editor check the field names. The package ships py.typed.

Next steps​

Last updated on

View this page as Markdown