pushary 1.1.0

Human-in-the-loop decisions SDK for Pushary: create a decision, ask a specific end-user to approve, and resume on their answer via webhook or poll.

pushary

Human-in-the-loop decisions for AI products, in Python. Create a decision, ask a specific end-user to approve it, and resume on their answer via webhook or poll.

This is the Python counterpart of the @pushary/server SDK. It is zero dependency (Python standard library only) and targets Python 3.9 and newer.

Installation

pip install pushary

API key

The SDK needs your full API key (pk_xxx.sk_xxx), which includes the secret half. Never expose it in client-side code. Get your key from your Pushary dashboard.

import os
from pushary import PusharyServer

pushary = PusharyServer(api_key=os.environ["PUSHARY_API_KEY"])

Human-in-the-loop decisions

A decision asks one of your end-users to approve something and then lets your product resume once they answer. Use it for the moments where a human should be in the loop: releasing funds, sending an outbound message, running a destructive action, or confirming an AI-proposed change.

The flow has three parts:

1. Create a decision and notify the end-user.
2. Resume when they answer, either by handling the webhook or by polling.
3. Optionally answer or cancel on their behalf from your own surface.

Create a decision

By default create is async: it returns right away with a decisionId and a pollUrl, which suits serverless functions that cannot hold a request open while a human decides. Pass wait=True to block for up to about 55 seconds in case the human answers quickly.

Always pass idempotency_key so a retried call does not ask the same human twice.

decision = pushary.decisions.create(
    "Approve the $4,200 payout to Acme Corp?",
    type="confirm",
    external_id="user_123",
    agent_name="Billing Agent",
    context="Invoice INV-8842, net-30, first payout to this vendor.",
    callback_url="https://your-app.com/webhooks/pushary",
    expires_in_seconds=3600,
    wait=True,
    timeout_seconds=50,
    idempotency_key="payout-INV-8842",
)

if decision["answered"]:
    print("Resolved fast:", decision["value"])
else:
    print("Still pending, poll:", decision["pollUrl"])

For a multiple choice decision, pass type="select" with at least two options. For free text, pass type="input" and an optional placeholder.

decision = pushary.decisions.create(
    "Which shipping speed should we book?",
    type="select",
    options=["Standard", "Express", "Overnight"],
    external_id="user_123",
)

Poll for the answer

If you did not wait, or the wait window closed before the human answered, poll for the outcome. Pass wait=N to long-poll for up to N seconds so the call returns as soon as they answer rather than on your next loop.

result = pushary.decisions.get(decision["decisionId"], wait=50)

if result["status"] == "answered":
    print("Answer:", result["value"])
elif result["status"] == "expired":
    print("The decision expired before anyone answered.")

Answer or cancel on their behalf

If your own interface collected the answer, record it so any waiting call resolves. Cancel a decision to close it when it is no longer needed.

pushary.decisions.answer(decision["decisionId"], "yes")

pushary.decisions.cancel(decision["decisionId"])

Webhooks

When the end-user answers, Pushary POSTs the result to your callback_url. The request carries an X-Pushary-Signature header, an HMAC-SHA256 hex digest of the raw request body signed with your webhook secret. Verify it against the raw bytes you received, before parsing the JSON, so a change to spacing or key order cannot slip past the check.

Fetch or rotate the secret from the SDK:

secret = pushary.decisions.get_webhook_secret()
rotated = pushary.decisions.rotate_webhook_secret()

A Flask handler that verifies the signature and resumes your work:

import os
from flask import Flask, request, abort
from pushary import verify_webhook_signature

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["PUSHARY_WEBHOOK_SECRET"]

@app.post("/webhooks/pushary")
def pushary_webhook():
    signature = request.headers.get("X-Pushary-Signature")
    if not verify_webhook_signature(request.data, signature, WEBHOOK_SECRET):
        abort(401)

    payload = request.get_json()
    decision_id = payload["decisionId"]
    answer = payload.get("value")
    # Resume your work now that the human has answered.
    return "", 204

request.data is the raw request body Flask captured, which is exactly what the signature was computed over. Do not re-serialize the parsed JSON before verifying.

Errors

Every method returns the parsed JSON body as a dict. A non-2xx response raises PusharyError, which carries the HTTP status and the reason the server reported.

from pushary import PusharyError

try:
    pushary.decisions.get("does-not-exist")
except PusharyError as error:
    print(error.status, error.message)

Security

• Keep your API key and webhook secret in environment variables.
• Rotate the webhook secret from the dashboard or via rotate_webhook_secret() if it is ever exposed.
• API keys are site-scoped, so a key can only reach its own decisions.

License

MIT. See LICENSE.