cloudpostoffice 1.2.3

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

cloudpostoffice

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

Install

pip install cloudpostoffice

Quick start

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

---

Direct Messages

Send a message directly from one postbox to another.

import asyncio
import cloudpostoffice as cpo

p1 = cpo.postbox('proj-xxx--postbox-1', 'your-secret')
p2 = cpo.postbox('proj-xxx--postbox-2', 'your-secret')

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

    # postbox-2 listens for incoming messages
    await p2.listen(on_message)

    # postbox-1 sends a message to postbox-2
    await p1.send(to='proj-xxx--postbox-2', msg='hello')

asyncio.run(main())

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

---

Pub/Sub

Any postbox 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

p1 = cpo.postbox('proj-xxx--postbox-1', 'your-secret')
p2 = cpo.postbox('proj-xxx--postbox-2', 'your-secret')

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

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

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

asyncio.run(main())

The subscribe callback receives (topic_name, message).

---

API

cpo.postbox(postbox_id, postbox_secret)

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

p = cpo.postbox('proj-xxx--postbox-1', 'my-secret')

---

await postbox.send(to, msg)

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

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

await p1.send(to='proj-xxx--postbox-2', msg='hello')

---

await postbox.listen(callback)

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

def on_message(msg):
    print(f"Message from {msg['from']}:", msg['msg'])

await p.listen(on_message)

---

await postbox.publish(topic_name, message)

Publishes a message to a named topic.

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

await p.publish('alerts', {'level': 'warn', 'text': 'High temp'})

---

await postbox.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 p.subscribe('alerts', on_alert)

---

postbox.disconnect()

Gracefully closes the MQTT connection.

p.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 clients cannot share the same postbox ID and secret at the same time within a project.

---

Tests

python tests\run_all.py

Links

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