Python client for Simplepush.
Send tasks, stream events over WebSocket, and decrypt end-to-end-encrypted payloads from Python.
pip install simplepush # HTTP + WebSocket only
pip install 'simplepush[crypto]'# adds end-to-end encryption supportRequires Python 3.10+.
fromsimplepushimportClient, TextInput, ChoiceInputclient=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 manyIds are type-prefixed strings.
task_id,subtask_id, input/reply/file ids and the like come back type-tagged —tsk_…,sub_…,inp_…,rpl_…(a reply),rfl_…(a reply's file) — not bare UUIDs.
By default every recipient gets their own independent task instance (one
recipient's answers never touch another's task), returned as a TaskGroup of
per-recipient Task handles:
group=client.send_task(topic="mytopic", content="check in")
fortaskingroup: # or group.instancesprint(task.task_id, task.recipient.public_id, task.recipient.name)
subs=group.append(content="follow-up") # a subtask on every member's chaingroup.append(content="just you", instances=[group.instances[0]]) # or a subsettask=client.send_task(topic="mytopic", content="hi", shared=True) # shared mode: ONE task everyone answers togetherBoth send methods take exactly one keyword-only target: topic= on any client,
or member= / broadcast= on an OrgClient. Omit the target on a personal
Client to send to your own devices (a self-send, returned as a single Task /
Notification; encrypted under the account personal password when one is
configured).
Other send options: auto_commit=False has the recipient submit the whole
form at once (by default each filled input arrives as an intermediate
InputEvent, then the terminal TaskCompleted carries the full committed
set); reply=ReplyMode.STICKY (or "one-shot" / "one-time-per-user") shows
recipients an in-thread reply composer (collect via replies());
content_format=ContentFormat.MARKDOWN renders content as Markdown;
critical=True sends an iOS Critical Alert.
A task can have subtasks appended to its chain. A subtask inherits the
parent's recipients and encryption (no target, no password); its inputs() /
replies() are scoped to it, and stream off the same shared connection:
sub=task.append(title="One more thing", inputs=[TextInput()])
asyncforevinsub.inputs():
ifisinstance(ev, SubtaskCompleted):
print(ev.uploads)Task inputs: TextInput, ChoiceInput (set multi=True, with optional
min_selections/max_selections), ActionsInput (styled buttons; the tapped
action's stable key comes back), SliderInput (min/max/step/unit),
PhotoInput, VoiceRecordingInput, FileUploadInput, LocationInput.
fromsimplepushimport (
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),
],
)
asyncforevinincident.inputs():
ifnotisinstance(ev.item, TaskCompleted):
continueforuinev.item.uploads:
matchu:
caseActionUpload(key=key):
print(ev.recipient.name, "pressed", key)
caseSliderUpload(value=value):
print("severity", value)
caseMultiChoiceUpload(values=values):
print("areas", values)
casePhotoUpload() asphoto:
awaitphoto.save("./incident-4711")Streams accept timeout= (seconds of silence before iteration stops; on a
group stream the timeout is group-wide) and replay=True (replay the buffered
backlog since the send before going live).
File downloads. The binary upload objects (photo/voice/file uploads, a
reply's photo/file/audio, and submission files) are download handles
bound to the client that yielded them: await x.read() returns the bytes
(checksum-verified, decrypted on encrypted chains), await x.save(path)
writes to disk (a directory uses the file's own name), and
await x.download_url() returns the raw short-lived presigned URL plus its
expiry. Failures raise DownloadError.
A notification is a lighter sibling of a task: it carries a single input (choice/text/actions only) and has no replies or subtasks.
Like send_task, the default is independent — every recipient gets their
own notification instance, returned as a NotificationGroup:
fromsimplepushimportClient, NotificationChoiceInput, NotificationActionInput, Action, ActionStyleclient=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 manyasyncforevinnote.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)A notification can also carry ONE media item — image= (renders on iOS +
Android) or audio= (plays inline on iOS only) — as either an http(s) URL or
a local file path (uploaded, encrypted when the notification is).
files= uploads local files alongside a task/subtask (encrypted when the send
is; each file is read fully into memory). A notification takes its single
media item the same way, or as a URL:
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")A submission is self-authored user content — a text body plus an optional
photo, file, audio clip, and location — pushed into a user's own stream with
no associated task; a task reply without the task. Submissions are created
by the app; the library observes them on the client's feed (both Client
and OrgClient):
asyncforsubinclient.submissions(timeout=300):
# sub: Submission — body / photo / file / audio / locationifsub.photo:
awaitsub.photo.save("./inbox")photo/file/audio are download handles (read() / save() /
download_url()); audio carries duration_seconds. location is inline
decoded data (latitude, longitude, accuracy, altitude, heading, speed,
timestamp). timeout= stops iteration after that many seconds of silence.
Encrypted submissions are decrypted with your personal password
(not a topic password). Pass it in passwords= (a bare string), or per call:
client=Client(api_token="USER_API_TOKEN", passwords="your-personal-password")
# or: client.submissions(password="your-personal-password")importasynciofromsimplepushimportClientasyncdefmain():
client=Client(api_token="USER_API_TOKEN")
asyncforeventinclient.events():
print(event.event_type, event.data)
asyncio.run(main())Every stream on a client shares one WebSocket. Call await client.aclose() when
you are done collecting; sends on their own never open it.
Pass password= to encrypt a send's body fields. The returned handle decrypts
the recipient's replies/inputs under the same password.
fromsimplepushimportClient, 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 sendasyncforreplyingroup.sole.replies():
print(reply.body) # decryptedTo decrypt the raw events() feed across many passwords, build a keyring from
the client's configured (password, topic) pairs (it also grows with every
send) and apply it per event:
fromsimplepushimporttry_decrypt_event_dataclient=Client(api_token="USER_API_TOKEN", passwords=[("hunter2", "mytopic"), ("other", "alerts")])
ring=client.keyring()
asyncforeventinclient.events():
data=try_decrypt_event_data(event, ring) # decrypted dict, or None if no key matchesOrgClient authenticates with the org api_key and addresses sends with
exactly one target: topic=, member= (by member name), or broadcast=True.
Encryption is automatic: pass the org's master key(s) (from your org's
encryption vault; the library can't derive them) and every send is encrypted
under the current key — there are no per-send passwords. Without keys, sends
go out in the clear and org ciphertext is passed through undecrypted.
fromsimplepushimportOrgClient, ChoiceInputorg=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"])],
)
asyncforevingroup.inputs():
print(ev.recipient.name, ev.item) # recipient = the org memberEverything else works as on a personal Client: independent-mode groups (the
member name rides on each instance's recipient), subtasks, streams,
submissions, downloads.
MIT