cloudpostoffice 1.2.1

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

Quick start

Each app or device needs a unique device ID and a secret key. Create them from your dashboard. Two apps cannot connect with the same device ID at the same time — every participant needs its own credentials.

---

Direct Messages

Send a message directly from one device to another.

import asyncio
import cloudpostoffice as cpo

d1 = cpo.device('device-1', 'your-secret')
d2 = cpo.device('device-2', 'your-secret')

async def main():
    def on_message(msg):
        print(msg)  # { 'from': 'device-1', 'msg': 'hello', 'ts': 1234567890 }

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

    # device-1 sends a message to device-2
    await d1.send(to='device-2', msg='hello')

asyncio.run(main())

The listen callback receives a dict { 'from', 'msg', 'ts' } where from is the sender's device ID, msg is the payload, and ts is the server timestamp.

---

Pub/Sub

Any device can publish or subscribe to any topic in the same project. No need to pre-create topics — they work on the fly.

import asyncio
import cloudpostoffice as cpo

d1 = cpo.device('device-1', 'your-secret')
d2 = cpo.device('device-2', 'your-secret')

async def main():
    def on_news(topic, msg):
        print(topic, msg)

    # d1 subscribes to a topic
    await d1.subscribe('news', on_news)

    # d2 publishes to the same topic
    await d2.publish('news', 'CloudPostOffice is alive!')

asyncio.run(main())

The subscribe callback receives (topic_name, message).

---

API

cpo.device(device_id, device_secret)

Creates a device handle. Automatically authenticates and connects to the MQTT broker on first use.

d = cpo.device('my-device', 'my-secret')

---

await device.send(to, msg)

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

Param	Type	Description
to	str	Target device ID
msg	any	Message payload (any JSON-serialisable value)

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

---

await device.listen(callback)

Registers a callback for messages addressed to this device. Can be called multiple times to add multiple handlers.

def on_message(msg):
    print(f"Message from {msg['from']}:", 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 is called whenever a message is published to that topic.

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

await d.subscribe('alerts', on_alert)

---

device.disconnect()

Gracefully closes the MQTT connection.

d.disconnect()

---

Notes

• Authentication tokens are valid for 7 days. The SDK will automatically reconnect and refresh the token when it expires.
• Topic names must not contain /, +, #, or --.
• Two devices cannot share the same device ID and secret at the same time within a project.

---

Links

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