This post documents necessary steps to connect an ESP32-DevKitC kit to Azure Iot Hub using MicroPython and the MQTT protocol. I am going to use the latest firmware released from MicroPython downloads.


This project requires the following MicroPython dependencies:

It is very simple to download and install dependencies through MicroPython’s upip package manager. If you load these modules at runtime, you are very likely to run into OOM issues when you don’t have external SPI PSRAM. What you can do is to freeze these modules into the firmware (Frozen Module) to save runtime compilation and memory.

MQTT Protocol

Since I can’t use the official Azure IoT Python SDK on ESP32, I have to connect to the public device endpoints using the MQTT protocol directly on port 8883. According to the Microsoft document, the device should use the following values in the CONNECT packet.

  • For the ClientId field, use the device_id.
  • For the Username field, use {hostname}/{device_id}/?api-version=2018-06-30, where {hostname} is the full CName of the IoT hub.
  • For the Password field, use a SAS token.

It is simple to parse device_id, hostname, and shared_access_key from the connection string. I am using Python string split calls. In production code, more validations are definitely required.

connection_string = ';DeviceId=MyIotDevice;SharedAccessKey=MySharedAccessKey'
parts = dict(x.split('=', 1) for x in connection_string.split(';'))

device_id = parts['DeviceId']
hostname = parts['HostName']
shared_access_key = parts['SharedAccessKey']


Assuming my IoT Hub name is and the name of my device is MyIotDevice, the Username field should be

username = '{}/{}/api-version=2018-06-30'.format(hostname, device_id)


For the Password field, there is more work involved in generating the SAS token. Luckily, after days of digging, I am able to find some hints from a long list of Azure Iot Hub developer guides. The security token has the following format: SharedAccessSignature sig={signature}&se={expiry}&skn={policyName}&sr={URL-encoded-resourceURI}.

Value Description
{signature} An HMAC-SHA256 signature string of the form: {URL-encoded-resourceURI} + "\n" + expiry. Important: The key is decoded from base64 and used as key to perform the HMAC-SHA256 computation.
{resourceURI} URI prefix (by segment) of the endpoints that can be accessed with this token, starting with host name of the IoT hub (no protocol).
{expiry} UTF8 strings for number of seconds since the epoch 00:00:00 UTC on 1 January 1970.
{URL-encoded-resourceURI} Lower case URL-encoding of the lower case resource URI.
{policyName} The name of the shared access policy to which this token refers. Absent if the token refers to device-registry credentials.
import hmac
import utime

from ubinascii import a2b_base64, b2a_base64
from uhashlib import sha256
from urllib.parse import quote_plus, urlencode

def get_time():
    # adjust timestamp on different ports
    t = utime.time()
    # embedded ports use epoch of 2000-01-01 00:00:00 UTC
    t += 946684800

    return t

def generate_sas_token(uri, key, policy_name=None, expiry=3600):
    ttl = int(get_time() + expiry)
    sign_key = '{}\n{}'.format(quote_plus(uri), ttl)
    signature = b2a_base64(, sign_key, sha256).digest())
    # strip off the trailing newline
    signature = signature[:-1]

    rawtoken = {
        'sr': uri,
        'sig': signature,
        'se': str(ttl)

    if policy_name:
        rawtoken['skn'] = policy_name

    return 'SharedAccessSignature {}'.format(urlencode(rawtoken))

uri = '{}/devices/{}'.format(hostname, device_id)
password = generate_sas_token(uri, shared_access_key)

Note that MicroPython embedded ports use epoch of 2000-01-01 00:00:00 UTC instead of the Unix epoch. 946684800 is the seconds between them.

Talk to the Cloud

With all the necessary information in hand, I am ready to connect to Azure Iot Hub using the micropython-umqtt.simple module.

from umqtt.simple import MQTTClient

def callback(topic, message):
    """Callback function for subscribed topics"""

    print('Received topic={} message={}'.format(topic, message))

mqttc = MQTTClient(
    device_id, hostname,
    user=username, password=password, ssl=True

# subscribe to cloud to device message
c2d_topic = 'devices/{}/messages/devicebound/#'.format(self.device_id)

# subscribe to direct method
dm_topic = '$iothub/methods/POST/#'

# subscribe to device twin
twin_topic = '$iothub/twin/res/#'

Error checking is omitted in the above snippet. For device-to-cloud (d2c) and cloud-to-device (c2d) communications, they are implemented as follows.

import ujson

def cloud_to_device():

def device_to_cloud():
    # device to cloud message
    d2c_topic = 'devices/{}/messages/events/'.format(device_id)
    mqttc.publish(d2c_topic, ujson.dumps(payload))

In order to check c2d messages every a few seconds and send d2c messages back to the cloud, I plugged everything into an asynchronous I/O loop. You can find my previous post on micropython-uasyncio for more details. If things are working correctly, you should be able to see d2c messages using the Azure CLI utility. When you send c2d messages or direct methods to the device, MQTT callback function will be triggered as well.