How do I use Python to publish MQTT messages to AWS IoT Core from my device?

8 minute read

I can't send or receive MQTT (Message Queuing Telemetry Transport) messages between AWS IoT Core and my device or MQTT client.

Short description

Verify that your AWS IoT thing is correctly configured and that its certificates are properly attached. To test your setup, you can use the AWS IoT MQTT client and the example Python code provided in this article.

Resolution

Set up a directory to test MQTT publishing

Create a working directory in your development environment. For example, create a directory named iot-test-publish.
Create a subdirectory for certificates in your new working directory. For example, create a subdirectory named certificates.
From the command line, change the directory to your new working directory.

Install pip and the AWS IoT SDK for Python

Note: Before you begin, install pip for Python 3 packaging. For more information, see Installation on the Python Packaging Authority (PyPA) website.

To install the AWS IoT SDK for Python v2, use the following command:
```
pip install awsiotsdk
```
-or-

To install the AWS IoT Device SDK for Python (the previous SDK version, v1), use the following command:
```
pip install AWSIoTPythonSDK
```

Note: It's a best practice to use the most recent update for your AWS IoT SDK for Python version. For more information, see AWS IoT SDK for Python v2 or AWS IoT Device SDK for Python on GitHub. It's also a best practice to use these SDKs to connect to AWS IoT Core, but these SDKs aren't required—you can use any other compliant third-party MQTT client to connect to AWS IoT Core.

Create an AWS IoT Core policy

Open the AWS IoT Core console.
In the left navigation pane, choose Secure.
Under Secure, choose Policies.
If you have existing AWS IoT Core policies, then choose Create to create a new policy.
-or-
If you don't have existing policies, then on the You don't have any policies yet page, choose Create a policy.
On the Create a policy page, enter a Name for your policy. For example, enter the name admin.
Under Add statements, complete the following actions:
For Action, enter iot:*.
Important: You can allow all AWS IoT actions (iot:*) for testing purposes. However, in a production setup, it's a best practice to increase security. For more secure policy examples, see AWS IoT Core policy examples.
For Resource ARN, enter *.
For Effect, select the Allow check box.
Choose Create.

For more information, see Create an AWS IoT policy and AWS IoT Core policies.

Create an AWS IoT thing

Note: You don't need to create a thing to connect to AWS IoT. However, things allow you to use additional security controls and other AWS IoT features, such as Fleet Indexing, Jobs, or Device Shadow.

In the AWS IoT Core console, in the left navigation pane, choose Manage.
If you have existing things, then choose Create to create a new thing.
-or-
If you don't have any existing things, then on the You don't have any things yet page, choose Register a thing.
On the Creating AWS IoT things page, choose Create a single thing.
On the Add your device to the thing registry page, complete the following actions:
Enter a Name for your thing. For example, enter Test-Thing.
(Optional) Under Add a type to this thing, choose or create a thing type.
(Optional) Under Add this thing to a group, choose or create a group. For more information about groups, see Static thing groups and Dynamic thing groups.
(Optional) Under Set searchable thing attributes (optional), add attributes as key-value pairs.
Choose Next.
Note: It’s a best practice to match the thing name to the clientId. If the names match, then you can use AWS IoT Core features like Fleet Indexing and Thing policy variables.
On the Add a certificate for your thing page, choose Create certificate. You see notifications that confirm that your thing and a certificate for your thing are created.
On the Certificate created page, complete the following actions:
Under In order to connect a device, you need to download the following, choose Download for the certificate, public key, and private key.
Save each downloaded file to the certificates subdirectory that you created earlier.
Under You also need to download a root CA for AWS IoT, choose Download. The Server authentication page opens to CA certificates for server authentication.
Under Amazon Trust Services Endpoints (preferred), choose Amazon Root CA 1. The certificate opens in your browser.
Copy the certificate (everything from -----BEGIN CERTIFICATE----- to -----END CERTIFICATE-----) and paste it into a text editor.
Save the certificate as a .pem file named root.pem to the certificates subdirectory.
On the Certificate created page in the AWS IoT Core console, choose Activate. The button changes to Deactivate.
Choose Attach a policy.
On the Add a policy for your thing page, complete the following actions:
Select the AWS IoT Core policy that you created earlier. For example, select admin.
Choose Register Thing.

For more information, see Create a thing object.

Copy the AWS IoT Core endpoint URL

In the AWS IoT Core console, in the navigation pane, choose Settings.
On the Settings page, under Custom endpoint, copy the Endpoint. This AWS IoT Core custom endpoint URL is unique to your AWS account and Region.

Create a Python program file

Save one of the following Python code examples as a Python program file named publish.py.

For AWS IoT SDK for Python v2, use the following example code:

Important: Replace customEndpointUrl with your AWS IoT Core custom endpoint URL. Replace certificates with the name of your certificates subdirectory. Replace a1b23cd45e-certificate.pem.crt with the name of your client .crt. Replace a1b23cd45e-private.pem.key with the name of your private key.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.# SPDX-License-Identifier: MIT-0

from awscrt import io, mqtt, auth, http
from awsiot import mqtt_connection_builder
import time as t
import json

# Define ENDPOINT, CLIENT_ID, PATH_TO_CERTIFICATE, PATH_TO_PRIVATE_KEY, PATH_TO_AMAZON_ROOT_CA_1, MESSAGE, TOPIC, and RANGE
ENDPOINT = "customEndpointUrl"
CLIENT_ID = "testDevice"
PATH_TO_CERTIFICATE = "certificates/a1b23cd45e-certificate.pem.crt"
PATH_TO_PRIVATE_KEY = "certificates/a1b23cd45e-private.pem.key"
PATH_TO_AMAZON_ROOT_CA_1 = "certificates/root.pem"
MESSAGE = "Hello World"
TOPIC = "test/testing"
RANGE = 20

# Spin up resources
event_loop_group = io.EventLoopGroup(1)
host_resolver = io.DefaultHostResolver(event_loop_group)
client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
mqtt_connection = mqtt_connection_builder.mtls_from_path(
            endpoint=ENDPOINT,
            cert_filepath=PATH_TO_CERTIFICATE,
            pri_key_filepath=PATH_TO_PRIVATE_KEY,
            client_bootstrap=client_bootstrap,
            ca_filepath=PATH_TO_AMAZON_ROOT_CA_1,
            client_id=CLIENT_ID,
            clean_session=False,
            keep_alive_secs=6
            )
print("Connecting to {} with client ID '{}'...".format(
        ENDPOINT, CLIENT_ID))
# Make the connect() call
connect_future = mqtt_connection.connect()
# Future.result() waits until a result is available
connect_future.result()
print("Connected!")
# Publish message to server desired number of times.
print('Begin Publish')
for i in range (RANGE):
    data = "{} [{}]".format(MESSAGE, i+1)
    message = {"message" : data}
    mqtt_connection.publish(topic=TOPIC, payload=json.dumps(message), qos=mqtt.QoS.AT_LEAST_ONCE)
    print("Published: '" + json.dumps(message) + "' to the topic: " + "'test/testing'")
    t.sleep(0.1)
print('Publish End')
disconnect_future = mqtt_connection.disconnect()
disconnect_future.result()

-or-

For AWS IoT Device SDK for Python v1, use the following example code:

Important: Replace customEndpointUrl with your AWS IoT Core custom endpoint URL. Replace certificates with the name of your certificates sub-directory. Replace a1b23cd45e-certificate.pem.crt with the name of your client .crt. Replace a1b23cd45e-private.pem.key with the name of your private key.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.# SPDX-License-Identifier: MIT-0

import time as t
import json
import AWSIoTPythonSDK.MQTTLib as AWSIoTPyMQTT

# Define ENDPOINT, CLIENT_ID, PATH_TO_CERTIFICATE, PATH_TO_PRIVATE_KEY, PATH_TO_AMAZON_ROOT_CA_1, MESSAGE, TOPIC, and RANGE
ENDPOINT = "customEndpointUrl"
CLIENT_ID = "testDevice"
PATH_TO_CERTIFICATE = "certificates/a1b23cd45e-certificate.pem.crt"
PATH_TO_PRIVATE_KEY = "certificates/a1b23cd45e-private.pem.key"
PATH_TO_AMAZON_ROOT_CA_1 = "certificates/root.pem"
MESSAGE = "Hello World"
TOPIC = "test/testing"
RANGE = 20

myAWSIoTMQTTClient = AWSIoTPyMQTT.AWSIoTMQTTClient(CLIENT_ID)
myAWSIoTMQTTClient.configureEndpoint(ENDPOINT, 8883)
myAWSIoTMQTTClient.configureCredentials(PATH_TO_AMAZON_ROOT_CA_1, PATH_TO_PRIVATE_KEY, PATH_TO_CERTIFICATE)

myAWSIoTMQTTClient.connect()
print('Begin Publish')
for i in range (RANGE):
    data = "{} [{}]".format(MESSAGE, i+1)
    message = {"message" : data}
    myAWSIoTMQTTClient.publish(TOPIC, json.dumps(message), 1) 
    print("Published: '" + json.dumps(message) + "' to the topic: " + "'test/testing'")
    t.sleep(0.1)
print('Publish End')
myAWSIoTMQTTClient.disconnect()

Note: Older versions of AWS IoT Device SDK for Python are not supported for Python 3.12 or later. Make sure that you use the most recent version of AWS IoT Device SDK for Python.

Test the setup

In the AWS IoT Core console, in the left navigation pane, choose Test.
On the MQTT client page, for Subscription topic, enter test/testing.
Choose Subscribe to topic. A test topic named test/testing is ready for test message publication. For more information, see View MQTT messages with the AWS IoT MQTT client.
Run the following command from your command line:
```
python3 publish.py
```

The Python program publishes 20 test messages to the topic test/testing that you created in the AWS IoT Core console. View the topic in the console to see the published messages.

Tip: You can also use the included pubsub examples to test other features of the SDKs, such as WebSockets subscriptions and connections. For more information, see pubsub (AWS IoT SDK for Python v2) or BasicPubSub (AWS IoT Device SDK for Python) on the GitHub website.

(Optional) Activate AWS IoT logging to Amazon CloudWatch

You can monitor event logs for MQTT messages that you publish to AWS IoT Core. For setup instructions, see Configure AWS IoT logging and Monitor AWS IoT using CloudWatch Logs.