IoTボタン × IoT Core × Lambda 連携でハマった落とし穴

以前、IoTボタンのクリックでS3に直接ファイルを保存する仕組みを構築しましたが、今回は受信したデータに対して何らかの処理を加えたいため、構成を「IoT Core → Lambda → S3」に変更しました。その際、予期せぬエラーに遭遇し、少しばかり時間を費やしてしまいました。本記事では、その時の状況と解決策を備忘録としてまとめます。

当初作成したLambda関数はこちらです。

import json

def lambda_handler(event, context):
    try:
        IoT Coreから送信されたJSONを受け取る
        payload = json.loads(event['body'])
        print(payload)


        # 項目を抽出
        

        return{
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Success'
            })
        }

    except Exception as e:
        print(f"Error: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Internal Server Error'
            })
        }

シンプルにLambdaでIoT Coreからの情報を受信するだけのコードを作成したつもりでした。受信した値から加工に必要な項目の抽出は後回しにし、動作テストを実施したところ、以下のエラーが発生しました。

Error: 'body'

Lambda関数が期待していたイベントの構造と、実際にIoT Coreから送信されたイベントの構造が異なっていたことが原因です。AWS IoT CoreからLambdaに送信されるイベントの構造は、HTTPリクエストのようなbodyを持つ構造ではないため、event['body'] にアクセスしようとするとエラーが発生します。実際にどのような構造になっているか確認するために、処理の冒頭に以下のようにログ出力処理を追加し、再度テストしてみました。

import json

def lambda_handler(event, context):
    print("イベント:", event) # ★★★ 追加 ★★★
    try:
        IoT Coreから送信されたJSONを受け取る
        payload = json.loads(event['body'])
        print(payload)


        # 項目を抽出
        

        return{
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Success'
            })
        }

    except Exception as e:
        print(f"Error: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Internal Server Error'
            })
        }

CloudWatch Logsで確認したイベントの内容から、IoTボタンの情報はイベントのルート直下にキーと値のペアで格納されていることが分かりました。

{'credentialsId': 'AWS-IAM-role-credentials-IoT-Publish', 'operatorId': 'OPXXXXXXXXXX', 'destination': {'provider': 'aws', 'service': 'aws-iot', 'resourceUrl': 'https://XXXXXXXXXX', 'payloadsOnly': False, 'sendPayloadsAsBinary': False}, 'sourceProtocol': 'udp', 'clickType': 1, 'clickTypeName': 'SINGLE', 'batteryLevel': 1, 'binaryParserEnabled': True, 'timestamp': 1744067245812, 'imsi': 'XXXXXXXXXX', 'imei': 'XXXXXXXXXX'}

今回の場合、直接ボタンの情報が渡されているため、それぞれの項目に直接アクセスすることが可能です。そのため、例えばIoT ボタンがどのようにクリックされたかを取得したい場合は、以下のコードで実現できました。

import json

def lambda_handler(event, context):
    try:

        # 項目を抽出
        clickTypeName = event['clickTypeName']
        print("クリックタイプ：", clickTypeName)


        return{
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Success'
            })
        }

    except Exception as e:
        print(f"Error: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Internal Server Error'
            })
        }