pip install simplepush            # HTTP + WebSocket only
pip install 'simplepush[crypto]'  # adds end-to-end encryption support

from simplepush import Client, TextInput, ChoiceInput
client = Client(api_token="USER_API_TOKEN")
group = client.send_task(
    topic="mytopic",
    title="Approve deploy?",
    inputs=[
        ChoiceInput(description="Deploy v1.2.3?", options=["yes", "no"], required=True),
        TextInput(description="Note (optional)"),
    ],
)
task = group.sole  # single-recipient topic; iterate the group for many

group = client.send_task(topic="mytopic", content="check in")
for task in group:      # or group.instances
    print(task.task_id, task.recipient.public_id, task.recipient.name)
subs = group.append(content="follow-up")                      # a subtask on every member's chain
group.append(content="just you", instances=[group.instances[0]])  # or a subset
task = client.send_task(topic="mytopic", content="hi", shared=True)  # shared mode: ONE task everyone answers together

sub = task.append(title="One more thing", inputs=[TextInput()])
async for ev in sub.inputs():
    if isinstance(ev, SubtaskCompleted):
        print(ev.uploads)

from simplepush import (
    Client, Action, ActionStyle, ActionsInput, SliderInput, ChoiceInput, PhotoInput,
    TaskCompleted, ActionUpload, SliderUpload, MultiChoiceUpload, PhotoUpload,
)
client = Client(api_token="USER_API_TOKEN")
incident = client.send_task(
    topic="ops",
    title="Incident 4711",
    inputs=[
        ActionsInput(actions=[
            Action(key="ack", label="Acknowledge", style=ActionStyle.PRIMARY),
            Action(key="escalate", label="Escalate", style=ActionStyle.DESTRUCTIVE),
        ]),
        SliderInput(min=0, max=10, step=1, unit="sev"),
        ChoiceInput(options=["db", "api", "infra"], multi=True, required=False),
        PhotoInput(required=False),
    ],
)
async for ev in incident.inputs():
    if not isinstance(ev.item, TaskCompleted):
        continue
    for u in ev.item.uploads:
        match u:
            case ActionUpload(key=key):
                print(ev.recipient.name, "pressed", key)
            case SliderUpload(value=value):
                print("severity", value)
            case MultiChoiceUpload(values=values):
                print("areas", values)
            case PhotoUpload() as photo:
                await photo.save("./incident-4711")

from simplepush import Client, NotificationChoiceInput, NotificationActionInput, Action, ActionStyle
client = Client(api_token="USER_API_TOKEN")
group = client.send_notification(
    topic="mytopic",
    title="Build failed",
    content="main @ a1b2c3 failed 3 tests",
    input=NotificationChoiceInput(options=["ack", "mute"]),
)
note = group.sole   # single-recipient topic; iterate the group for many
async for ev in note.inputs():
    print(ev.reply)   # NotificationTextReply / NotificationChoiceReply / NotificationActionReply
# Action buttons (approve/deny), like a task's ActionsInput — on an encrypted
# send both the `key` and the `label` are sealed, and so is the reported answer:
group = client.send_notification(
    topic="mytopic",
    title="Deploy v1.2.3?",
    input=NotificationActionInput(actions=[
        Action(key="approve", label="Approve"),
        Action(key="deny", label="Deny", style=ActionStyle.DESTRUCTIVE),
    ]),
)
# Shared mode: ONE notification all recipients see and answer together (the
# first answer completes it for everyone), returned as a plain `Notification`:
note = client.send_notification(topic="mytopic", content="heads up", shared=True)

client.send_task(
    topic="reports",
    title="Q3 numbers",
    content="Full report attached.",
    files=["q3.pdf"],
)
client.send_notification(topic="alerts", title="Door cam", image="https://cam.example/last.jpg")

async for sub in client.submissions(timeout=300):
    # sub: Submission — body / photo / file / audio / location
    if sub.photo:
        await sub.photo.save("./inbox")

client = Client(api_token="USER_API_TOKEN", passwords="your-personal-password")
# or: client.submissions(password="your-personal-password")

import asyncio
from simplepush import Client
async def main():
    client = Client(api_token="USER_API_TOKEN")
    async for event in client.events():
        print(event.event_type, event.data)
asyncio.run(main())

from simplepush import Client, ReplyMode
# Per-send password (`reply=` so there is a composer to collect from):
client = Client(api_token="USER_API_TOKEN")
group = client.send_task(topic="mytopic", title="Secret", content="🤫",
                         password="hunter2", reply=ReplyMode.STICKY)
# Or configure a topic's password on the client; sends to it omit `password=`,
# and a per-send password still overrides. The pair's topic must match the
# topic you send to — otherwise nothing matches and the send goes plaintext:
client = Client(api_token="USER_API_TOKEN", passwords=[("hunter2", "mytopic")])
client.send_task(topic="mytopic", content="🤫")              # encrypted with "hunter2"
client.send_task(topic="mytopic", content="!", password="x") # overridden for this send
async for reply in group.sole.replies():
    print(reply.body)   # decrypted

from simplepush import try_decrypt_event_data
client = Client(api_token="USER_API_TOKEN", passwords=[("hunter2", "mytopic"), ("other", "alerts")])
ring = client.keyring()
async for event in client.events():
    data = try_decrypt_event_data(event, ring)   # decrypted dict, or None if no key matches

from simplepush import OrgClient, ChoiceInput
org = OrgClient(
    api_key="ORG_API_KEY",
    master_key=MASTER_KEY,   # 32 bytes (raw or base64)
    master_key_version=3,    # or several: master_keys={3: key3, 2: key2}
)
group = org.send_task(
    broadcast=True,
    title="All hands?",
    inputs=[ChoiceInput(options=["yes", "no"])],
)
async for ev in group.inputs():
    print(ev.recipient.name, ev.item)   # recipient = the org member