AMQP to Azure IoTHub via CAT-M

  • Hi Pycom folks,
    I'd like to use the FiPy (LTE CAT-M pycom) to upload data to Azure IoTHub. I can find bits of pieces of what's needed, but I'm unsure about how to string these together.

    In order to push data to Azure IoTHub over CAT-M, a few things are needed:

    1. SHA256 HMAC for authentication to Azure
    2. TLS and NTP and network
    3. AMQP (I want AMQP instead HTTPS to save on data usage)

    Here are the pieces I've found.

    For #1 and #2, Dave Glover from Microsoft has an Azure client implementation for micropython targeted for ESP32 (awesome!). However, it uses wlan interface built into the ESP32, and he uses HTTPS instead of AMQP to publish to IoTHub.

    My question is: Does it matter that socket is using wlan or LTE? Can I pretty much just replace his wlan lines with the LTE lines from the pycom example? A socket is a socket regardless, and the network library abstracts wlan and LTE for me?

    //esp32 wifi interface
    wlan = network.WLAN(network.STA_IF)
    s = socket.socket()

    Where as for the LTE connection on pycom, it looks like this

    //pycom example for LTE interface
    s = socket.socket()

    So with that, I think I can get an HTTP example to Azure IoTHub.

    As for #3, converting the HTTP example to AMQP. I couldn't find any micropython AMQP libraries at all. There are some popular pure python AMQP libraries, but being new to micropython, I'm not sure what would be involved to convert python to micropython.

  • Banned

    This post is deleted!

  • Banned

    This post is deleted!

  • @shp96 I am using TLS with AWS Mqtt so TLS does work, with a local public certificate. Can't comment if the AMQP lib supports it.

  • @seb Hi, thanks for your input.

    I have been trying to communicate a SIPY with the Azure IOT hub.

    I have used the same libraries that you use and the syntax is the same but the server gives an empty answer and the board becomes unable to stablish a connection.

    I've done tests with brokers without security like Hive's and everything works without inconvenience. As it seems the problem begins when implementing SSL / TLS, because by default the ESP32 firmware does not work with TLSv1 and the azure platform requires TLSv1.

    Did you manage to communicate your device with the azure platform and visualize the messages that you sent in the device explore?

    Also I'm sure that the problem isn't the SAS because I've stablished MQTT connections using the MQTTfx windows client with TLSv1.

  • @seb I think the SAS Token has changed, or I was doing something wrong, I rewrote the generate_sas_token code and it now works for me.

    from base64 import b64encode, b64decode
    from hashlib import sha256
    from time import time
    from hmac import HMAC
    from urllib.parse import quote_plus, urlencode, quote
    def generate_sas_token(uri, key, keyName, ttl=3600):
        expiry = int(time() + ttl)
        uri = quote_plus(uri)
        sas = key.encode('utf-8')
        string_to_sign = (uri + '\n' + str(expiry)).encode('utf-8')
        signed_hmac_sha256 = HMAC(sas, string_to_sign, sha256)
        signature = quote(b64encode(signed_hmac_sha256.digest()))
        return {
            'expiry': expiry,
            'token': 'SharedAccessSignature sr={}&sig={}&se={}&skn={}'.format(uri, signature, expiry, keyName)

    It looks a little different to yours, though to be honest this hasn't been tested on my LoPy yet, I just wanted to get it working in Python using micropython libraries.

    Leaving the code here for completeness should someone return in the future.

    Also, here is my eventhub file that I'm using to send events to Event Hub.
    import urequests
    import azure_sas
    from time import time as seconds 
    class EventHub:
        def __init__(self, options):
            self.endpoint = options['endpoint'] # Endpoint without post URL on the end. From Connection String 
            self.keyname = options['keyname'] # Endpoint Key Name. From Connection String 
            self.key = options['key'] # Endpoint Key, From Connection String 
            self.url = options['url'] # Complete POST URL for messages
            self.token = None 
            self.tokenExpires = None
        def make_token(self):
            Returns an authorization token dictionary 
            for making calls to Event Hubs REST API.
            if self.token is None or self.tokenExpires < seconds():
                token_gen = azure_sas.generate_sas_token(self.endpoint, self.key, self.keyname)
                self.tokenExpires = token_gen['expiry']
                self.token = token_gen['token']
        def publish_single(self, payload, userProperties=None, brokerProperties=None):
            if userProperties is not None or brokerProperties is not None:
                encapsulatedPayload = {
                    'Body': payload
                if userProperties is not None:
                    encapsulatedPayload['UserProperties'] = userProperties
                if brokerProperties is not None: 
                    encapsulatedPayload['BrokerProperties'] = brokerProperties
                return self.publish_multiple([encapsulatedPayload])
            return self.publish_multiple([{'Body': payload}])
        def publish_multiple(self, payloads):
            headers = {
                'Content-Type': 'application/',
                'Authorization': self.token,
                'x-ms-retrypolicy': 'NoRetry'
            uri = self.url + '?timeout=60&api-version=2014-01'
                import json as ujson
                import ujson
            payload = ujson.dumps(payloads).encode('utf-8')
            print('EventBus Publish')
            return, headers=headers, data=payload)

    I should also mention that the endpoint I'm using is https://{servicebus}{hubname}. The Connection String uses sb:// and that didn't seem to work for me.

  • @nevercast

    It took me quite a while to figure out how to generate the sas tokens so Im happy to share my code, Microsoft's documentation on the topic is quite cryptic.

  • Thank you for this seb! The timing couldn't have been better. I am doing something similar with the Event Hub and this covers the SAS Token which was not working with the modules on board. Thanks.

  • Hi,

    I recently managed to connect to azure IoT-Hub using MQTT (sorry not AMQP) using the following code:

    from base64 import b64encode, b64decode
    from sha256 import sha256
    from time import time
    from hmac import HMAC
    from parse import quote_plus, urlencode, quote
    from network import WLAN
    from mqtt import MQTTClient
    import machine
    print("connecting to wifi")
    wlan = WLAN(mode=WLAN.STA)
    wlan.connect("SSID", auth=(WLAN.WPA2, "PASSWORD"), timeout=5000)
    while not wlan.isconnected():
    print("Connected to Wifi\n")
    # Set time
    print("Setting time")
    rtc = machine.RTC()
    while not rtc.synced():
    print("Time set to: {}".format(
    def generate_sas_token(uri, key, policy_name="iothubowner", expiry=None):
        uri = quote(uri, safe='').lower()
        encoded_uri = quote(uri, safe='').lower()
        if expiry is None:
            expiry = time() + 3600
        ttl = int(expiry)
        sign_key = '%s\n%d' % (encoded_uri, ttl)
        signature = b64encode(HMAC(b64decode(key), sign_key.encode('utf-8'), sha256).digest())
        result = 'SharedAccessSignature ' + urlencode({
            'sr': uri,
            'sig': signature,
            'se': str(ttl),
            'skn': policy_name
        return result
    # Azure IoT-hub settings
    hostname = ""
    #This needs to be the key of the "device" IoT-hub shared access policy, NOT the device
    policy_name = "device"
    primary_key = "XXXXXXXXXXXXXX"
    device_id = "seb-test-2"
    uri = "{hostname}/devices/{device_id}".format(hostname=hostname, device_id=device_id)
    password = generate_sas_token(uri, primary_key, policy_name)
    username_fmt = "{}/{}/api-version=2016-11-14"
    username = username_fmt.format(hostname, device_id)
    client = MQTTClient(device_id, hostname, user=username, password=password,
                        ssl=True, port=8883)
    topic = "devices/{device_id}/messages/events/".format(device_id=device_id)
    client.publish(topic, "test", qos=1)

    You will need to download a bunch of libraries from here:

    • base64
    • collections.defaultdict
    • hmac
    • hashlib (the one in the firmware doesnt work for this purpose)
    • urllib.parse
    • warnings


Pycom on Twitter