hey-pingr 0.2.0

Official Python SDK for the Hey Pingr WhatsApp auto-reply platform

pingr (Python)

Official Python SDK for Hey Pingr — WhatsApp auto-reply automation.

Hey Pingr works by listening for incoming trigger phrases on your connected WhatsApp number and automatically sending your configured reply. Your server receives a signed webhook event for every inbound message and can optionally return a dynamic reply — the text Hey Pingr will send to the user instead of the static auto-reply.

Installation

pip install hey-pingr

How it works

1. A user sends a trigger phrase (e.g. "login") to your WhatsApp number
2. Hey Pingr instantly sends either the static auto-reply you configured — or the dynamic reply your webhook server returns
3. Hey Pingr POSTs a signed message.received webhook event to your server so you can act on it

Auth patterns

Pattern 1 — Mobile-first (simplest)

from flask import Flask, request, jsonify
from pingr import verify_webhook, PingrWebhookError
import os

app = Flask(__name__)

@app.route("/webhook", methods=["POST"])
def webhook():
    try:
        payload = verify_webhook(
            request.get_data(),
            request.headers["x-pingr-signature"],
            os.environ["PINGR_WEBHOOK_SECRET"],
        )
        if payload["event"] == "message.received":
            phone = payload["from"]["phone"]
            link  = generate_magic_link(phone)
            # Return dynamic reply — Hey Pingr sends this as the WhatsApp message
            return jsonify({"reply": f"Your login link: {link}"})
        return "", 200
    except PingrWebhookError:
        return "", 400

Pattern 2 — Phone pre-entry + browser polling

import redis
from flask import Flask, request, jsonify
from pingr import verify_webhook

r = redis.Redis()

@app.route("/webhook", methods=["POST"])
def webhook():
    payload = verify_webhook(request.get_data(), request.headers["x-pingr-signature"], SECRET)
    if payload["event"] == "message.received":
        phone = payload["from"]["phone"]
        link  = generate_magic_link(phone)
        r.set(f"auth:{phone}", link, ex=300)
        return jsonify({"reply": f"Tap to sign in: {link}"})
    return "", 200

@app.route("/auth/poll")
def poll():
    phone = request.args.get("phone")
    link  = r.get(f"auth:{phone}")
    if link:
        r.delete(f"auth:{phone}")
        return jsonify({"status": "ready", "link": link.decode()})
    return jsonify({"status": "pending"})

Pattern 3 — Desktop challenge code (no phone pre-entry)

from pingr import verify_webhook, create_challenge_code, extract_challenge_code

pending = {}  # use Redis in production

@app.route("/auth/challenge")
def challenge():
    code = create_challenge_code()  # e.g. "X4K9MQ"
    pending[code] = {"status": "pending"}
    return jsonify({"code": code, "trigger_phrase": f"login {code}"})

@app.route("/webhook", methods=["POST"])
def webhook():
    payload = verify_webhook(request.get_data(), request.headers["x-pingr-signature"], SECRET)
    if payload["event"] == "message.received":
        code = extract_challenge_code(payload["message"]["text"], "login")
        if code and pending.get(code, {}).get("status") == "pending":
            link = generate_magic_link(payload["from"]["phone"])
            pending[code] = {"status": "ready", "link": link}
            return jsonify({"reply": f"Here's your link: {link}"})
    return "", 200

@app.route("/auth/poll")
def poll():
    code = request.args.get("code")
    s    = pending.get(code)
    if not s:
        return jsonify({"status": "not_found"}), 404
    if s["status"] == "pending":
        return jsonify({"status": "pending"})
    link = pending.pop(code)["link"]
    return jsonify({"status": "ready", "link": link})

Dynamic reply

Any 2xx JSON response your webhook returns with a reply string field will be sent as the WhatsApp message to the user — replacing the static auto-reply.

return jsonify({"reply": "Your magic link: https://yourapp.com/auth?token=..."})

• Maximum 4096 characters (WhatsApp limit)
• Non-JSON or missing reply → static auto-reply is used (fully backward-compatible)

FastAPI example

from fastapi import FastAPI, Request, HTTPException
from pingr import verify_webhook, PingrWebhookError
import os

app = FastAPI()

@app.post("/webhook")
async def webhook(request: Request):
    raw_body  = await request.body()
    signature = request.headers.get("x-pingr-signature", "")
    try:
        payload = verify_webhook(raw_body, signature, os.environ["PINGR_WEBHOOK_SECRET"])
    except PingrWebhookError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    if payload["event"] == "message.received":
        phone = payload["from"]["phone"]
        link  = await generate_magic_link(phone)
        return {"reply": f"Your login link: {link}"}

    return {"ok": True}

Webhook payloads

message.received

{
  "event": "message.received",
  "session_id": "sess_abc123",
  "from": {
    "phone": "919876543210",
    "jid": "[e-mail]",
    "is_group": false,
    "group_jid": null
  },
  "message": { "text": "login X4K9MQ", "type": "text" },
  "timestamp": 1714825320000
}

session.disconnected

{
  "event": "session.disconnected",
  "session_id": "sess_abc123",
  "reason": 401,
  "kind": "terminal",
  "timestamp": 1714825320000
}

API reference

verify_webhook(raw_body, signature, secret, *, check_timestamp=True)

Verify the x-pingr-signature header and return the parsed payload dict.
Raises PingrWebhookError on failure.

Param	Type	Description
raw_body	bytes / str	Raw request body — do not JSON-parse first
signature	str	Value of the x-pingr-signature header
secret	str	Your webhook signing secret from the dashboard
check_timestamp	bool	Reject payloads older than 5 min. Default: True

create_challenge_code(*, length=6, charset=...)

Generate a cryptographically random challenge code using Python's secrets module.

code = create_challenge_code()        # "X4K9MQ"
code = create_challenge_code(length=8) # "X4K9MQAB"

extract_challenge_code(message_text, trigger_phrase)

Parse the challenge code suffix from a trigger message.

extract_challenge_code("login X4K9MQ", "login")  # → "X4K9MQ"
extract_challenge_code("login", "login")          # → None
extract_challenge_code("help", "login")           # → None

License

MIT