or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

device-shadows.mdexception-handling.mdgreengrass-discovery.mdindex.mdiot-jobs.mdmqtt-client.md
tile.json

tessl/pypi-awsiotpythonsdk

SDK for connecting to AWS IoT using Python.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/awsiotpythonsdk@1.5.x

To install, run

npx @tessl/cli install tessl/pypi-awsiotpythonsdk@1.5.0

index.mddocs/

AWS IoT Python SDK

A comprehensive Python SDK for connecting devices to AWS IoT platform through MQTT and MQTT over WebSocket protocols. Enables developers to build IoT applications that can securely communicate with AWS IoT services, manage device shadows (digital representations of device state), handle AWS IoT Jobs for device management, and integrate with AWS Greengrass for edge computing scenarios.

Package Information

  • Package Name: AWSIoTPythonSDK
  • Language: Python
  • Installation: pip install AWSIoTPythonSDK

Core Imports

import AWSIoTPythonSDK.MQTTLib as AWSIoTPyMQTT

For Greengrass discovery:

from AWSIoTPythonSDK.core.greengrass.discovery.providers import DiscoveryInfoProvider

For exception handling:

from AWSIoTPythonSDK.exception.AWSIoTExceptions import connectTimeoutException
from AWSIoTPythonSDK.exception.operationTimeoutException import operationTimeoutException
from AWSIoTPythonSDK.exception.operationError import operationError

For IoT Jobs:

from AWSIoTPythonSDK.core.jobs.thingJobManager import jobExecutionTopicType
from AWSIoTPythonSDK.core.jobs.thingJobManager import jobExecutionTopicReplyType
from AWSIoTPythonSDK.core.jobs.thingJobManager import jobExecutionStatus

For Device Shadow operations:

from AWSIoTPythonSDK.core.shadow.deviceShadow import deviceShadow

Basic Usage

TLS Mutual Authentication

import AWSIoTPythonSDK.MQTTLib as AWSIoTPyMQTT

# Create MQTT client
myAWSIoTMQTTClient = AWSIoTPyMQTT.AWSIoTMQTTClient("myClientID")

# Configure endpoint and credentials  
myAWSIoTMQTTClient.configureEndpoint("your-endpoint.iot.region.amazonaws.com", 8883)
myAWSIoTMQTTClient.configureCredentials("rootCA.crt", "private.key", "certificate.crt")

# Configure auto-reconnect
myAWSIoTMQTTClient.configureAutoReconnectBackoffTime(1, 32, 20)

# Connect and publish
myAWSIoTMQTTClient.connect()
myAWSIoTMQTTClient.publish("myTopic", "Hello from AWS IoT device", 0)
myAWSIoTMQTTClient.disconnect()

WebSocket with SigV4

import AWSIoTPythonSDK.MQTTLib as AWSIoTPyMQTT

# Create WebSocket client
myAWSIoTMQTTClient = AWSIoTPyMQTT.AWSIoTMQTTClient("myClientID", useWebsocket=True)

# Configure endpoint and IAM credentials
myAWSIoTMQTTClient.configureEndpoint("your-endpoint.iot.region.amazonaws.com", 443)
myAWSIoTMQTTClient.configureIAMCredentials("accessKeyID", "secretAccessKey", "sessionToken")

# Connect and subscribe
myAWSIoTMQTTClient.connect()
myAWSIoTMQTTClient.subscribe("myTopic/sub", 1, customCallback)

Architecture

The SDK is built around three main client types that share a common configuration pattern:

  • AWSIoTMQTTClient: Core MQTT client providing direct publish/subscribe operations
  • AWSIoTMQTTShadowClient: Specialized client for device shadow operations with persistent subscriptions
  • AWSIoTMQTTThingJobsClient: Specialized client for AWS IoT Jobs management and execution

All clients support both TLS mutual authentication and WebSocket SigV4 authentication, with automatic reconnection, progressive backoff, and offline request queueing capabilities.

Capabilities

Core MQTT Operations

Basic MQTT connectivity including connection management, publish/subscribe operations, and advanced features like auto-reconnect, offline queueing, and custom authentication methods.

class AWSIoTMQTTClient:
    def __init__(self, clientID: str, protocolType: int = 4, useWebsocket: bool = False, cleanSession: bool = True): ...
    def configureEndpoint(self, hostName: str, portNumber: int): ...
    def configureCredentials(self, CAFilePath: str, KeyPath: str = "", CertificatePath: str = "", Ciphers: str = None): ...
    def connect(self, keepAliveIntervalSecond: int = 600) -> bool: ...
    def publish(self, topic: str, payload: str, QoS: int) -> bool: ...
    def subscribe(self, topic: str, QoS: int, callback: callable) -> bool: ...

Core MQTT Operations

Device Shadow Management

Device shadow operations for synchronizing device state between physical devices and the cloud, including shadow retrieval, updates, deletion, and delta notifications for state changes.

class AWSIoTMQTTShadowClient:
    def __init__(self, clientID: str, protocolType: int = 4, useWebsocket: bool = False, cleanSession: bool = True, awsIoTMQTTClient = None): ...
    def createShadowHandlerWithName(self, shadowName: str, isPersistentSubscribe: bool) -> 'deviceShadow': ...

class deviceShadow:
    def shadowGet(self, callback: callable, timeout: int) -> str: ...
    def shadowUpdate(self, JSONPayload: str, callback: callable, timeout: int) -> str: ...
    def shadowDelete(self, callback: callable, timeout: int) -> str: ...

Device Shadow Management

AWS IoT Jobs

Job execution management for device fleet management, including job subscription, execution state management, progress reporting, and job lifecycle operations.

class AWSIoTMQTTThingJobsClient:
    def __init__(self, clientID: str, thingName: str, QoS: int = 0, protocolType: int = 4, useWebsocket: bool = False, cleanSession: bool = True, awsIoTMQTTClient = None): ...
    def sendJobsStartNext(self, statusDetails: dict = None, stepTimeoutInMinutes: int = None) -> bool: ...
    def sendJobsUpdate(self, jobId: str, status: int, statusDetails: dict = None, expectedVersion: int = 0, executionNumber: int = 0, includeJobExecutionState: bool = False, includeJobDocument: bool = False, stepTimeoutInMinutes: int = None) -> bool: ...
    def createJobSubscription(self, callback: callable, jobExecutionType = None, jobReplyType = None, jobId: str = None) -> bool: ...

AWS IoT Jobs

Greengrass Discovery

Discovery service for finding and connecting to AWS Greengrass Core devices in local networks, including connectivity information retrieval and CA certificate management.

class DiscoveryInfoProvider:
    def __init__(self, caPath: str = "", certPath: str = "", keyPath: str = "", host: str = "", port: int = 8443, timeoutSec: int = 120): ...
    def configureEndpoint(self, host: str, port: int = 8443): ...
    def configureCredentials(self, caPath: str, certPath: str, keyPath: str): ...
    def discover(self, thingName: str) -> 'DiscoveryInfo': ...

Greengrass Discovery

Exception Handling

Comprehensive exception classes for timeout conditions, operation errors, connection failures, and Greengrass discovery issues, enabling robust error handling in IoT applications.

# Timeout exceptions
class connectTimeoutException(Exception): ...
class publishTimeoutException(Exception): ...
class subscribeTimeoutException(Exception): ...

# Operation errors  
class connectError(Exception): ...
class publishError(Exception): ...
class subscribeError(Exception): ...

# Discovery errors
class DiscoveryTimeoutException(Exception): ...
class DiscoveryUnauthorizedException(Exception): ...

Exception Handling

Constants and Enums

The SDK provides various constants and enumerations for configuration and operation:

# SDK version information
import AWSIoTPythonSDK
__version__ = "1.5.4"  # Available as AWSIoTPythonSDK.__version__

# MQTT Protocol versions
MQTTv3_1 = 3
MQTTv3_1_1 = 4

# Queue drop behavior
DROP_OLDEST = 0
DROP_NEWEST = 1

# Job execution status values
class jobExecutionStatus:
    JOB_EXECUTION_STATUS_NOT_SET = (0, None)
    JOB_EXECUTION_QUEUED = (1, 'QUEUED')
    JOB_EXECUTION_IN_PROGRESS = (2, 'IN_PROGRESS')
    JOB_EXECUTION_FAILED = (3, 'FAILED')
    JOB_EXECUTION_SUCCEEDED = (4, 'SUCCEEDED')
    JOB_EXECUTION_CANCELED = (5, 'CANCELED')
    JOB_EXECUTION_REJECTED = (6, 'REJECTED')
    JOB_EXECUTION_UNKNOWN_STATUS = (99, None)