AWS Common Runtime (awscrt)

Warnings

• breaking `os.fork()` is unsafe when used with `awscrt` due to its use of background threads. In a forked child process, background threads vanish, potentially leading to hangs or crashes when the child attempts to communicate with them. This impacts the default behavior of Python's `multiprocessing` module on POSIX systems (except macOS) in Python versions 3.13 and earlier.
  Fix: When using `multiprocessing`, set the start method to 'spawn' or 'forkserver' (e.g., `multiprocessing.set_start_method('spawn')`). If `fork` must be used, ensure all CRT resources are released and all CRT threads are joined before forking (e.g., via `awscrt.common.join_all_native_threads()`).
• gotcha On macOS, once a private key is used with a certificate, that certificate-key pair is imported into the Mac Keychain. All subsequent uses of that certificate will use the stored private key from the Keychain and ignore any private key passed in programmatically. This can lead to unexpected behavior if you intend to use different private keys for the same certificate.
  Fix: Be aware of this platform-specific behavior. If you need to use a different private key, you may need to manage certificates in the Keychain or use distinct certificates. awscrt logs an 'info' level message when a key from the Keychain is used instead of a provided one.
• gotcha When building `awscrt` from source on Unix systems, `s2n-tls` is used, which depends on `libcrypto` (from OpenSSL). By default, `awscrt` includes its own statically compiled copy of `libcrypto` from AWS-LC. If you explicitly need `awscrt` to use a `libcrypto` included on your system, you must set the `AWS_CRT_BUILD_USE_SYSTEM_LIBCRYPTO=1` environment variable during installation and use `--no-binary :all:`, which can complicate the build process.
  Fix: If you require the system's `libcrypto`, set `AWS_CRT_BUILD_USE_SYSTEM_LIBCRYPTO=1` and use `pip install --no-binary :all: awscrt`. Otherwise, rely on the default build process which bundles its own `libcrypto` for simpler installation.

Install

• pip install awscrt Install from PyPI

Imports

• io

  from awscrt import io

  Core I/O primitives like EventLoopGroup, HostResolver, and SocketOptions.
• mqtt

  from awscrt import mqtt

  MQTT client functionalities for connecting to AWS IoT Core.
• auth

  from awscrt import auth

  AWS client-side authentication including credential providers and signing.
• http

  from awscrt import http

  HTTP client and related functionalities, used for WebSocket connections and proxies.
• mqtt_connection_builder

  from awsiot import mqtt_connection_builder

  mqtt_connection_builder is part of the 'awsiot' package, which builds upon awscrt, not directly from awscrt itself.

Quickstart

This quickstart demonstrates how to establish an MQTT connection to AWS IoT Core and publish a message using `awscrt`. It sets up basic I/O resources, builds an mTLS connection, connects, publishes a message, and then disconnects. Environment variables are used for sensitive credentials and endpoint configuration.

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

# Replace with your AWS IoT Core endpoint, client ID, and certificate paths
ENDPOINT = os.environ.get("AWS_IOT_ENDPOINT", "YOUR_AWS_IOT_ENDPOINT")
CLIENT_ID = "testDevice"
PATH_TO_CERTIFICATE = os.environ.get("AWS_IOT_CERT_PATH", "certificates/certificate.pem.crt")
PATH_TO_PRIVATE_KEY = os.environ.get("AWS_IOT_PRIVATE_KEY_PATH", "certificates/private.pem.key")
PATH_TO_AMAZON_ROOT_CA_1 = os.environ.get("AWS_IOT_ROOT_CA_PATH", "certificates/root-CA.pem")
TOPIC = "test/topic"
MESSAGE = "Hello from awscrt Python!"

# 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,
    ca_filepath=PATH_TO_AMAZON_ROOT_CA_1,
    client_bootstrap=client_bootstrap,
    client_id=CLIENT_ID,
    clean_session=False,
    keep_alive_secs=30
)

print(f"Connecting to {ENDPOINT} with client ID '{CLIENT_ID}'...")
connect_future = mqtt_connection.connect()
connect_future.result()
print("Connected!")

print(f"Publishing message to topic '{TOPIC}': {MESSAGE}")
mqtt_connection.publish(topic=TOPIC, payload=MESSAGE, qos=mqtt.QoS.AT_LEAST_ONCE).result()
print("Published!")

# Disconnect
disconnect_future = mqtt_connection.disconnect()
disconnect_future.result()
print("Disconnected!")