cloudpostoffice 1.2.0

Python SDK for CloudPostOffice — messaging for AI agents, apps, and devices

cloudpostoffice

Python SDK for CloudPostOffice — super-simple messaging for AI agents, apps, and devices.

Install

pip install cloudpostoffice

Or from source:

git clone https://github.com/CloudPostOffice/python
cd python
pip install .

Requirements: Python 3.10+

---

Quick start

Each app or device needs a unique device ID and secret key. Create them from your dashboard.

---

Direct Messages

Send a message directly from one device to another.

import asyncio
import cloudpostoffice as cpo

cpo.configure(base_url="https://cloudpostoffice.com")

d1 = cpo.device("device-1-id", "device-1-secret")
d2 = cpo.device("device-2-id", "device-2-secret")

async def main():
    def on_message(msg):
        print(msg)
    # msg is { "from": "device-1-id", "msg": "hello", "ts": 1234567890000 }

    # device-2 listens for incoming messages
    await d2.listen(on_message)

    # device-1 sends to device-2
    await d1.send(to="device-2-id", msg="hello")
    d1.disconnect()

asyncio.run(main())

---

Pub / Sub

Any device can publish or subscribe to a named topic within the same project.

import asyncio
import cloudpostoffice as cpo

cpo.configure(base_url="https://cloudpostoffice.com")

async def main():
    publisher  = cpo.device("device-1-id", "device-1-secret")
    subscriber = cpo.device("device-2-id", "device-2-secret")

    def on_news(topic, msg):
        print(topic, "->", msg)

    # subscriber listens first — always subscribe before publishing
    await subscriber.subscribe("news", on_news)
    print("Subscribed, waiting for messages...")

    # publisher sends
    await publisher.publish("news", "CloudPostOffice is alive!")
    publisher.disconnect()

    # keep the event loop alive to receive messages
    await asyncio.Event().wait()

asyncio.run(main())

---

Decorator-based client (direct MQTT, no auth)

CloudPostOfficeClient connects directly to any MQTT broker without the CloudPostOffice authentication layer. Useful for prototyping or when you already have broker credentials.

import asyncio
from cloudpostoffice import CloudPostOfficeClient

cpo = CloudPostOfficeClient("broker.emqx.io")

@cpo.on("cloudpostoffice/inbox/user123")
def on_inbox(payload: str) -> None:
    print(f"[inbox] {payload}")

@cpo.on("cloudpostoffice/alerts/#")
async def on_alert(payload: str) -> None:
    print(f"[alert] {payload}")

asyncio.run(cpo.listen())

Both def and async def handlers are accepted transparently.

---

API

cpo.configure(*, base_url)

Override the API base URL. Call before creating any devices.

cpo.configure(base_url="https://cloudpostoffice.com")

---

cpo.device(device_id, device_secret) → Device

Creates a device handle. Authenticates and connects to the MQTT broker automatically on first use.

d = cpo.device("my-device-id", "my-secret")

---

await device.send(*, to, msg)

Sends a direct message to another device on the same project.

Param	Type	Description
to	str	Target device ID
msg	any	Payload — any JSON-serialisable value

await d1.send(to="device-2-id", msg="hello")

---

await device.listen(callback)

Registers a callback for messages addressed to this device.

Callback signature: fn(message) — message is {"from", "msg", "ts"}.

def on_message(msg):
    print(msg["from"], "says:", msg["msg"])

await d.listen(on_message)

---

await device.publish(topic_name, message)

Publishes a message to a named topic.

• Topic names must not contain /, +, #, or --.

await d.publish("alerts", {"level": "warn", "text": "High temp"})

---

await device.subscribe(topic_name, callback)

Subscribes to a named topic.
Callback signature: fn(topic_name, message).

def on_alert(topic, msg):
    print(topic, msg)

await d.subscribe("alerts", on_alert)

---

device.disconnect()

Signals the background connection task to stop.

d.disconnect()

---

CloudPostOfficeClient(broker_url, port=1883)

Lightweight decorator-based MQTT client for direct broker connections.

from cloudpostoffice import CloudPostOfficeClient

cpo = CloudPostOfficeClient("broker.emqx.io")

@cpo.on("topic/#")
async def handler(payload: str) -> None:
    print(payload)

asyncio.run(cpo.listen())

---

Running the tests

1. Fill in credentials

Edit .env.test:

CPO_BASE_URL=https://cloudpostoffice.com
CPO_TEST_DEVICE_1_ID=your-device-1-id
CPO_TEST_DEVICE_1_SECRET=your-device-1-secret
CPO_TEST_DEVICE_2_ID=your-device-2-id
CPO_TEST_DEVICE_2_SECRET=your-device-2-secret

Create two devices from your dashboard.

2. Install dependencies

pip install -r requirements.txt

3. Run all tests

python tests/run_all.py

Expected output:

CloudPostOffice Python SDK — Integration Tests
Base URL  : https://cloudpostoffice.com
Device 1  : your-device-1-id
Device 2  : your-device-2-id

Pub / Sub (topic bus)
---------------------
  PASS  pub_test.py — published successfully
  PASS  sub_test.py — received message on correct topic

Device send / listen
--------------------
  PASS  device1.py — message sent
  PASS  device2.py — message received

Security — unauthorized cross-project publish
---------------------------------------------
  PASS  unauth_test.py — broker rejected unauthorized publish

Passed: 5   Failed: 0

Unit tests (no credentials needed)

python tests/test_client_id_split.py

---

Notes

• Authentication tokens are valid for 7 days. The SDK reconnects and refreshes automatically when a token expires.
• Topic names must not contain /, +, #, or --.
• Two devices cannot share the same device ID simultaneously within a project.
• On Windows, the SDK automatically switches to WindowsSelectorEventLoopPolicy so aiomqtt works correctly (paho-mqtt requires add_reader/add_writer which are not available on the default ProactorEventLoop).
• Requires Python 3.10+.

---

Links

• Dashboard
• Documentation
• Email: [e-mail]