Python SDK

Installation

pip install simplepush

Python 3.10+. For end-to-end encryption, install the crypto extra, which pulls in PyNaCl:

pip install "simplepush[crypto]"

Without the extra, plaintext sending and receiving work normally. Encrypted payloads pass through as ciphertext.

Clients

The SDK has two clients:

• Client: a personal account, authenticated with your API token (app settings, under API Token).
• OrgClient: an organization, authenticated with the organization API key.

Sending is synchronous. send_task and send_notification are plain method calls that return a handle immediately. Collecting answers is asynchronous. The handle's streams are async iterators that you consume with async for.

from simplepush import Client, OrgClient

client = Client(api_token="YOUR_API_TOKEN")
org = OrgClient(api_key="YOUR_ORG_API_KEY")

Constructor options

Parameter	Client	OrgClient	Description
host, port, ssl	✓	✓	Server override, defaults api.simplepu.sh / 443 / True.
api_token	required	—	Your personal API token.
passwords	✓	—	Encryption passwords, see below.
api_key	—	required	The organization API key.
master_keys	—	✓	Organization encryption keys as a {version: key} map (or master_key= plus master_key_version= for a single key).

passwords takes either a single string or a list:

client = Client(
    api_token="YOUR_API_TOKEN",
    passwords=[
        ("alerts-secret", "alerts"),   # (password, topic): encrypts sends to that topic and decrypts its events
        "personal-secret",              # bare string: your personal password
    ],
)

A (password, topic) pair becomes the default key for that topic, for sending and receiving. Only one bare string is allowed: the personal password. It decrypts your submissions and encrypts what you send to your own devices. It never encrypts a topic send.

Sending tasks

send_task takes exactly one target: topic= on both clients, or member= or broadcast=True on OrgClient only. With no target, a personal Client sends to your own devices and returns a single Task.

from simplepush import Client, ChoiceInput

client = Client(api_token="YOUR_API_TOKEN")

group = client.send_task(
    topic="deploys",
    title="Approve deploy?",
    inputs=[ChoiceInput(options=["Approve", "Deny", "Hold"])],
)

By default every recipient gets their own task instance, and the send returns a TaskGroup. When there is exactly one recipient, group.sole gives you that single Task. Pass shared=True to send one task that all recipients see and answer together. That send returns a plain Task.

Parameter	Type	Description
topic / member / broadcast	targeting	Exactly one, or none to send to your own devices.
title, content	str | None	Task title and body. At least content or inputs is required.
inputs	list[Input] | None	Inputs the recipient should fill, see Input types.
links	list[str] | None	URL attachments.
files	list[str | PathLike] | None	Local files to upload and attach. Encrypted with the send's key before upload.
auto_commit	bool	Complete the task automatically once all required inputs are filled. Default True.
password	str | None	Password to encrypt this send with. Requires topic=. Defaults to the topic's configured password, if there is one. Otherwise the send is unencrypted. Not accepted when sending to your own devices. Those sends always use your personal password.
reply	ReplyMode | str | None	Show a reply composer: "one-shot", "sticky", or "one-time-per-user". Omit for none.
content_format	str | None	"markdown" renders content as Markdown on the recipient's device. The format marker is never encrypted. Default plain.
tag	str | None	Label for receiver-side filtering.
shared	bool	One shared task for all recipients instead of one each, see above. Default False.
expires_at	datetime | str | None	Deadline. A timezone-aware datetime or an ISO 8601 string, in the future. After the deadline, an unanswered task expires. See Declining and expiry.

Returns a TaskGroup (default), or a Task when shared=True or the send goes to your own devices.

Input types

All accept description= and required= (default True).

Class	Use
TextInput(default_value=…)	Free-form text reply.
ChoiceInput(options=[…])	One-of-many choice. Add multi=True for multi-select, optionally with min_selections= / max_selections=. A multi-select never renders as push buttons in notifications. The recipient answers in the app.
ActionsInput(actions=[Action(key=…, label=…, style=…)])	Buttons the recipient taps. The chosen key comes back. style is "default", "primary", or "destructive", or an ActionStyle. Keys must be unique.
SliderInput(min=…, max=…, step=…, unit=…, default_value=…)	A number on a [min, max] scale. min / max required, the rest optional.
PhotoInput()	Photo from the camera.
VoiceRecordingInput()	Voice recording.
FileUploadInput()	Arbitrary file upload.
LocationInput()	GPS location from the device.

Collecting answers

Handles expose async streams. inputs() yields each filled input as an InputEvent. The stream ends with one terminal marker: TaskCompleted (carrying all uploads), TaskDeleted, TaskCanceled, TaskDeclined, or TaskExpired. replies() yields Reply objects from the task's reply composer. Both accept timeout=, the seconds of silence after which the stream ends (None waits forever), and replay=True, which also delivers the backlog since the send.

import asyncio
from simplepush import Client, ChoiceInput, TaskCompleted

client = Client(api_token="YOUR_API_TOKEN")

group = client.send_task(
    topic="deploys",
    title="Approve deploy?",
    inputs=[ChoiceInput(options=["Approve", "Deny"])],
)

async def main():
    async for g in group.inputs(timeout=600):
        if isinstance(g.item, TaskCompleted):
            print(g.recipient, "answered", g.item.uploads)

asyncio.run(main())

A TaskGroup streams all recipients at once. group.inputs() and group.replies() yield GroupInput and GroupReply wrappers with .instance (that recipient's own Task), .item, and .recipient. timeout measures silence across the whole group, so one quiet recipient does not end the stream.

The upload objects are typed: TextUpload(value), ChoiceUpload(index, value), MultiChoiceUpload(indices, values), ActionUpload(key), SliderUpload(value), LocationUpload(location), and the downloadable PhotoUpload / VoiceUpload / FileUpload (see Downloads).

Subtasks

Append a follow-up to a task the recipient already has. task.append(...) returns a Subtask with its own inputs() / replies() streams. group.append(...) appends to every instance and returns a list. Pass instances=[…] to append to some instances only. Subtasks accept the same content, inputs, links, and files as send_task. They go one level deep only.

sub = task.append(content="One more thing: confirm the version", inputs=[TextInput()])

Canceling

Every send handle can withdraw what it sent. task.cancel(...) and sub.cancel(...) return once the server has accepted the cancel. group.cancel(...) cancels every still-pending instance and returns a GroupCancelResult with the counts. Already-finished instances are skipped, never failed.

result = group.cancel(reason="answered", note="already handled")
print(result.canceled, result.skipped)

Options:

• reason: a CancelReason enum member or its string value. "canceled" (default), "answered", or "superseded".
• note: free text for the recipients. Encrypted under the chain's key when the send was encrypted.
• superseded_by: the replacement, as an id or a handle. Requires reason="superseded". A subtask's replacement must be in the same chain. A group cancel takes the replacement group, and each instance is pointed at its own recipient's replacement.

On the receiving side, a canceled instance's streams end with a TaskCanceled marker. It carries reason, note, and superseded_by. A canceled subtask yields SubtaskCanceled on its own streams only. A canceled root task ends every stream of its chain. Canceling an already-answered task raises ApiError with code == "task_already_completed". Canceling twice is a no-op.

Declining and expiry

An instance's streams can also end because of a decline or an expiry. Both arrive as typed markers, like TaskCanceled.

When a recipient declines in the app, the stream yields a TaskDeclinedByRecipient signal. A shared task stays open for the other recipients. The signal carries reason ("declined" or "failed"), an optional note, and an actor that says who declined. When every recipient has declined, the terminal TaskDeclined marker follows and the stream ends. An independent-mode instance has one recipient, so its signal and terminal marker arrive together. Subtask declines work the same way with SubtaskDeclinedByRecipient / SubtaskDeclined, and affect only that subtask.

A task sent with expires_at= expires when its deadline passes without an answer. Its streams end with a TaskExpired marker. Like TaskDeclined, this marker has no actor and no note, because the deadline was set on the send. An expired or fully-declined root task ends every stream of its chain, exactly like a canceled one.

async for g in group.inputs():
    match g.item:
        case TaskDeclinedByRecipient(reason=r, note=n):
            print(g.recipient, "declined:", r, n)
        case TaskExpired():
            print(g.recipient, "never answered in time")

Late answers into a declined or expired task raise ApiError with code == "task_declined" / "task_expired".

Sending notifications

send_notification uses the same targeting and grouping as send_task, but sends a lighter, fire-and-forget push. It carries at most one input and at most one media item. The media item is either image= or audio=, never both. Each takes a URL string or a local path to upload. Notification inputs have their own types: NotificationTextInput(), NotificationChoiceInput(options=[…]), or NotificationActionInput(actions=[…]).

from simplepush import Client, NotificationChoiceInput

client = Client(api_token="YOUR_API_TOKEN")

group = client.send_notification(
    topic="deploys",
    title="Restart the gateway?",
    content="Staging is wedged",
    input=NotificationChoiceInput(options=["Yes", "No"]),
)

async def main():
    async for done in group.sole.inputs(timeout=300):
        print(done.reply)  # NotificationChoiceReply(selected_index=0, selected_value="Yes")

A notification handle has a single stream. inputs() yields exactly one NotificationCompleted, with the reply if there is one, and then ends. A group has a single stream too. group.inputs() merges every instance's stream and yields GroupNotification wrappers with .instance, .item, and .recipient. It ends once every recipient has answered. timeout measures silence across the whole group.

Media constraints: images (image/jpeg, image/png, image/gif) render on iOS and Android. Audio plays inline on iOS only.

Submissions

Submissions are content a person sends from the app without being asked: text, a photo, a voice memo, a file, a location. Observe them with client.submissions():

async def main():
    async for s in client.submissions():
        if s.body:
            print("text:", s.body.text)
        if s.photo:
            await s.photo.save("./inbox")

Encrypted submissions are encrypted under your personal password. Configure it in passwords, or pass password= per call, to decrypt them. OrgClient.submissions() works the same way with the organization master keys.

Downloads

Photo, voice, and file objects yielded by any stream are bound download handles with three async methods:

Method	Returns
await x.read()	The bytes, checksum-verified and decrypted.
await x.save(path=None)	Writes to disk (a directory uses the file's own name) and returns the path.
await x.download_url()	(presigned_url, expires_at), valid for about 5 minutes. Use it to fetch with your own HTTP stack. On an encrypted chain the URL serves the raw ciphertext, so prefer read() / save(), which verify and decrypt.

Raw events

client.events() yields every raw Event on your account's stream, not split per send. Use it for manual inspection or custom routing. For normal use, prefer the handle streams above.

Errors

Exception	Raised when
ApiError(status, body)	A send or HTTP call fails with a non-2xx response.
StreamError	The WebSocket fails and cannot recover. Short drops reconnect silently.
DownloadError	A download fails, or a handle is used outside a client stream.

Cleanup

Call await client.aclose() when you're done collecting to close the shared WebSocket. Sends alone don't need it.