Metadata-Version: 2.4
Name: cloudpostoffice
Version: 1.2.0
Summary: Python SDK for CloudPostOffice — messaging for AI agents, apps, and devices
Home-page: https://github.com/CloudPostOffice/python
Author: Kishore
Author-email: Kishore <hi@cloudpostoffice.com>
License: ISC
Project-URL: Homepage, https://cloudpostoffice.com
Project-URL: Repository, https://github.com/CloudPostOffice/python
Project-URL: Issues, https://github.com/CloudPostOffice/python/issues
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: ISC License (ISCL)
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Communications
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aiomqtt>=2.0
Requires-Dist: aiohttp>=3.9
Dynamic: author
Dynamic: home-page
Dynamic: license-file
Dynamic: requires-python

# cloudpostoffice

Python SDK for [CloudPostOffice](https://cloudpostoffice.com) — super-simple messaging for AI agents, apps, and devices.

## Install

```bash
pip install cloudpostoffice
```

Or from source:

```bash
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](https://cloudpostoffice.com/app).

---

## Direct Messages

Send a message directly from one device to another.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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 |

```python
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"}`.

```python
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 `--`.

```python
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)`.

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

await d.subscribe("alerts", on_alert)
```

---

### `device.disconnect()`

Signals the background connection task to stop.

```python
d.disconnect()
```

---

### `CloudPostOfficeClient(broker_url, port=1883)`

Lightweight decorator-based MQTT client for direct broker connections.

```python
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](https://cloudpostoffice.com/app).

### 2. Install dependencies

```bash
pip install -r requirements.txt
```

### 3. Run all tests

```bash
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)

```bash
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](https://cloudpostoffice.com/app)
- [Documentation](https://cloudpostoffice.com/docs)
- Email: [hi@cloudpostoffice.com](mailto:hi@cloudpostoffice.com)
