Python quickstart

Connect a Python service to CloudSignal in five minutes using paho-mqtt or aiomqtt.

Connect a Python service to CloudSignal using any standard MQTT 5 client. Use paho-mqtt for synchronous code and aiomqtt for asyncio-based services. Both authenticate against the same broker with the same username@org_id credentials.

Install

pip install paho-mqtt

pip install aiomqtt

Connect

The Python clients connect over TLS on port 8883. Authenticate with the username format your_username@your_organization_id and the password you set when creating the client. Find your organization ID in Settings → Organization in the dashboard.

import os
import ssl
import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, reason_code, properties):
    print('connected')
    client.subscribe('agents/agent-01/state')

def on_message(client, userdata, msg):
    print(f'{msg.topic}: {msg.payload.decode()}')

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id='py-agent-01')
client.username_pw_set(os.environ['MQTT_USERNAME'], os.environ['MQTT_PASSWORD'])
client.tls_set(cert_reqs=ssl.CERT_REQUIRED)

client.on_connect = on_connect
client.on_message = on_message

client.connect('mqtt.cloudsignal.app', 8883)
client.loop_forever()

import asyncio
import os
import ssl
import aiomqtt

async def main():
    async with aiomqtt.Client(
        hostname='mqtt.cloudsignal.app',
        port=8883,
        username=os.environ['MQTT_USERNAME'],
        password=os.environ['MQTT_PASSWORD'],
        tls_context=ssl.create_default_context(),
        identifier='py-agent-01',
    ) as client:
        await client.subscribe('agents/agent-01/state')

        async for message in client.messages:
            print(f'{message.topic}: {message.payload.decode()}')

asyncio.run(main())

Set the credentials in a .env file or your secret manager. Username is the full username@organization_id string.

# .env (never commit this)
MQTT_USERNAME=agent-01@a1b2c3d4-e5f6-7890-abcd-ef1234567890
MQTT_PASSWORD=your-password

Publish

Once connected, publish to any topic the client has ACL permission for. Payloads can be bytes or strings; serialize JSON yourself.

import json

client.publish(
    'agents/agent-01/state',
    json.dumps({'status': 'idle', 'load': 0.12}),
    qos=1,
)

import json

await client.publish(
    'agents/agent-01/state',
    payload=json.dumps({'status': 'idle', 'load': 0.12}),
    qos=1,
)

Authenticate via token exchange

For browser-mediated flows, exchange an identity provider token (Supabase, Firebase, Auth0, Clerk) for temporary MQTT credentials before connecting. See the Token exchange API.

import os
import requests

response = requests.post(
    'https://api.cloudsignal.io/v2/tokens/exchange',
    headers={
        'Authorization': f"Bearer {os.environ['CLOUDSIGNAL_SECRET_KEY']}",
        'Content-Type': 'application/json',
    },
    json={
        'provider': 'supabase',
        'token': supabase_access_token,
    },
)
credentials = response.json()
# → { "mqtt_username": "...", "mqtt_password": "...", "expires_at": "..." }

Then connect paho-mqtt or aiomqtt with credentials['mqtt_username'] and credentials['mqtt_password']. Refresh before expires_at.

Never embed the CloudSignal secret key in client-side code. Run the exchange call from your backend and pass the returned MQTT credentials to the client.

Last will

To publish a status message automatically if your client disconnects unexpectedly, set a last-will message before connecting.

client.will_set(
    'agents/agent-01/state',
    payload='{"status": "offline"}',
    qos=1,
    retain=True,
)

The broker publishes this payload to subscribers on the topic when it detects an unclean disconnect.

Next steps

SDK catalog - Other recommended clients for Go, Rust, Java, and .NET
Connection parameters - Hosts, ports, and TLS reference
ACL rules - Limit which topics this Python service can publish or subscribe to
Token exchange API - Full reference for the exchange endpoint