AWS IoT Greengrass AI Agents Context Packがリリース！AI エージェントでデバイスアプリケーションを開発してみた

$ lsb_release -a
No LSB modules are available.
Distributor ID:	Ubuntu
Description:	Ubuntu 24.04.2 LTS
Release:	24.04
Codename:	noble

$ uname -a
Linux ip-10-60-4-20 6.14.0-1014-aws #14~24.04.1-Ubuntu SMP Tue Sep 23 14:51:14 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux

cd ~
mkdir -p amazon-q-install
cd amazon-q-install

curl --proto '=https' --tlsv1.2 -sSf \
"https://desktop-release.q.us-east-1.amazonaws.com/latest/q-x86_64-linux.zip" \
-o "q.zip"

$ ls -lh q.zip
-rw-rw-r-- 1 ubuntu ubuntu 196M Nov  5 09:03 q.zip

sudo apt install unzip

$ unzip q.zip
Archive:  q.zip
   creating: q/
  inflating: q/install.sh
  inflating: q/README
  inflating: q/BUILD-INFO
   creating: q/bin/
  inflating: q/bin/q
  inflating: q/bin/qterm
  inflating: q/bin/qchat

$ ./q/install.sh
✔ Do you want q to modify your shell config (you will have to manually do this otherwise)? · Yes
✔ Select login method · Use for Free with Builder ID

Confirm the following code in the browser
Code: XXXX-XXXX

Open this URL: https://view.awsapps.com/start/#/device?user_code=XXXX-XXXX
▰▰▱▱▱▱▱ Logging in...

$ git clone https://github.com/aws-greengrass/greengrass-agent-context-pack.git
cd greengrass-agent-context-pack

$ q chat

/agent create --name greengrass-agent

{
	"$schema": "https://raw.githubusercontent.com/aws/amazon-q-developer-cli/refs/heads/main/schemas/agent-v1.json",
	"name": "greengrass-agent",
	"description": "AWS IoT Greengrass development assistant with context pack",
	"prompt": "You are an expert AWS IoT Greengrass developer assistant. You have access to comprehensive Greengrass setup guides,component examples, and deployment workflows. Always follow the documentation-first workflow and verify end-to-end functionality.",
	"tools": [
		"fs_read",
		"fs_write",
		"execute_bash",
		"use_aws"
	],
	"allowedTools": [
		"fs_read"
	],
	"toolsSettings": {
		"fs_read": {
			"allowedPaths": [
				"~/greengrass-agent-context-pack/**"
			]
		},
		"execute_bash": {
			"allowReadOnly": true,
			"allowedCommands": [
				"git status",
				"docker ps",
				"docker images",
				"aws iot .*",
				"aws greengrass .*"
			]
		},
		"use_aws": {
			"allowedServices": [
				"iot",
				"greengrass",
				"s3",
				"iam",
				"logs"
			]
		}
	},
	"resources": [
		"file://~/greengrass-agent-context-pack/README.md",
		"file://~/greengrass-agent-context-pack/AGENTS.md",
		"file://~/greengrass-agent-context-pack/setup/**/*.md",
		"file://~/greengrass-agent-context-pack/components/**/*.yaml",
		"file://~/greengrass-agent-context-pack/examples/**/*.md"
	],
	"hooks": {
		"agentSpawn": [
			{
				"command": "echo 'Working directory: '$(pwd)",
				"timeout_ms": 5000
			}
		]
	}
}

/quit

$ q chat --agent greengrass-agent

- 次のダミーデータを生成して、デバイスのローカルファイルにデータを記録します。
  - timestamp: データの生成時間です
  - 整数のダミーデータ：5秒ごとにインクリメントした値です。
  - ファイルはCSV形式でカラムは `timestamp,data` にしてください。
- 上記のデータを `/tmp/ggc-dummy.csv` というファイルに追記形式で出力します。
- カスタムコンポーネントの名前は `agent-test-local-component` にしてください
- カスタムコンポーネントは python で作成してください

#!/usr/bin/env python3

import csv
import time
from datetime import datetime

def main():
    counter = 0
    csv_file = '/tmp/ggc-dummy.csv'

    # CSVヘッダーを書き込み（ファイルが存在しない場合のみ）
    try:
        with open(csv_file, 'x', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['timestamp', 'data'])
    except FileExistsError:
        pass

    while True:
        timestamp = datetime.now().isoformat()

        # CSVファイルにデータを追記
        with open(csv_file, 'a', newline='') as f:
            writer = csv.writer(f)
            writer.writerow([timestamp, counter])

        counter += 1
        time.sleep(5)

if __name__ == "__main__":
    main()

run: 'python3 {artifacts:decompressedPath}/agent-test-local-component/src/main.py'

/quit

$ q chat --agent greengrass-agent

次の 2 種類のセンサーデータをダミーで生成して、AWS IoT Core に MQTT で 5 秒間隔で Publish するカスタムコンポーネントを作成してください。

- value: 整数で 5 秒ごとにインクリメントした値
- timestamp: データ生成した時間を ISO8601 形式で記載

パブリッシュ先のトピックは、`topic/agent-test` にしてください。
データのフォーマットは　JSON にしてください。
カスタムコンポーネントの名前は `agent-test-ipc` にしてください
カスタムコンポーネントは python で作成してください

#!/usr/bin/env python3

import json
import logging
import os
import sys
import time
from datetime import datetime, timezone

try:
    from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2
    GREENGRASS_IPC_AVAILABLE = True
except ImportError:
    GREENGRASS_IPC_AVAILABLE = False
    logging.warning("Greengrass IPC が利用できません - シミュレーションモードで実行")

# ログ設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('AgentTestIPC')

class AgentTestPublisher:
    def __init__(self):
        self.config = self.load_configuration()
        self.ipc_client = None
        self.counter = 0
        self.setup_ipc_client()

    def load_configuration(self):
        """コンポーネント設定を読み込み"""
        try:
            config = {
                "topic": "topic/agent-test",
                "interval": 5
            }

            # 環境変数から設定を読み込み（テスト用）
            config["topic"] = os.environ.get('GG_TOPIC', config["topic"])
            config["interval"] = int(os.environ.get('GG_INTERVAL', config["interval"]))

            return config
        except Exception as e:
            logger.error(f"設定の読み込みに失敗: {e}")
            raise

    def setup_ipc_client(self):
        """Greengrass IPC V2クライアントを初期化"""
        if GREENGRASS_IPC_AVAILABLE:
            try:
                self.ipc_client = GreengrassCoreIPCClientV2()
                logger.info("Greengrass IPC V2に接続しました")
            except Exception as e:
                logger.error(f"Greengrass IPC V2への接続に失敗: {e}")
                self.ipc_client = None
        else:
            logger.info("シミュレーションモードで実行 - メッセージはログのみに出力")

    def generate_sensor_data(self):
        """センサーデータを生成"""
        self.counter += 1

        data = {
            "value": self.counter,
            "timestamp": datetime.now(timezone.utc).isoformat()
        }

        return data

    def publish_to_iot_core(self, message_data):
        """IoT Coreにメッセージを送信"""
        try:
            message_json = json.dumps(message_data)

            if self.ipc_client:
                # 実際のGreengrass環境 - IPC V2使用
                self.ipc_client.publish_to_iot_core(
                    topic_name=self.config['topic'],
                    payload=message_json.encode('utf-8'),
                    qos=1
                )

                logger.info(f"IoT Coreトピック '{self.config['topic']}' に送信: {message_json}")
            else:
                # シミュレーションモード
                logger.info(f"[シミュレーション] トピック '{self.config['topic']}' に送信予定: {message_json}")

        except Exception as e:
            logger.error(f"メッセージの送信に失敗: {e}")
            raise

    def run(self):
        """メインコンポーネントループ"""
        logger.info("Agent Test IPCコンポーネントを開始...")
        logger.info(f"設定: {json.dumps(self.config, indent=2)}")

        try:
            while True:
                sensor_data = self.generate_sensor_data()
                self.publish_to_iot_core(sensor_data)
                time.sleep(self.config['interval'])

        except KeyboardInterrupt:
            logger.info("Agent Test IPCコンポーネントを停止...")
        except Exception as e:
            logger.error(f"予期しないエラー: {e}")
            sys.exit(1)
        finally:
            if self.ipc_client:
                self.ipc_client.close()

if __name__ == "__main__":
    publisher = AgentTestPublisher()
    publisher.run()

#!/usr/bin/env python3

import json
import logging
import time
import traceback
from datetime import datetime, timezone

from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2

# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def main():
    counter = 0

    try:
        # IPC クライアント作成
        ipc_client = GreengrassCoreIPCClientV2()
        logger.info("IPC クライアントに接続しました")

        while True:
            counter += 1

            # メッセージ作成
            message = {
                "value": counter,
                "timestamp": datetime.now(timezone.utc).isoformat()
            }

            # IoT Core に送信 (payloadをbytes型に変換)
            ipc_client.publish_to_iot_core(
                topic_name="topic/agent-test",
                payload=json.dumps(message).encode('utf-8'),
                qos=1
            )

            logger.info(f"メッセージ送信: {message}")

            # 5秒待機
            time.sleep(5)

    except Exception as e:
        logger.error(f"エラー: {e}")
        logger.error(traceback.format_exc())

if __name__ == "__main__":
    main()

エラーになっています。次の点を修正してください。

- レシピの「"run": "python3 {artifacts:path}/src/main.py"」を「run: 'python3 {artifacts:decompressedPath}/agent-test-ipc/src/main.py'」に修正してデプロイして下さい。
- IPCクライアントは 新しいバージョンである V2 を使ってください
	- 参考ドキュメント：https://docs.aws.amazon.com/greengrass/v2/developerguide/interprocess-communication.html#ipc-service-connect
- Pythonのコードは下記のドキュメントに記載のコードを参考にして修正して下さい
	 - https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-iot-core-mqtt.html
- Greengrassデバイスの Python環境には awsiotsdk がインストールされていないので、レシピでインストールする指示を追加してください。実行するのは `python3 -m pip install awsiotsdk` です。
- ファイルの修正が終わったら、デプロイを実行せず変更内容を出力してください。

- 35    :                 payload=json.dumps(message),
+ 35    :                 payload=json.dumps(message).encode('utf-8'),

# Greengrass デバイス上で手動で実行
sudo apt install python3-pip

  "Manifests": [
    {
      "Platform": {
        "os": "linux"
      },
      "Lifecycle": {
        "run": "python3 {artifacts:path}/src/main.py"
      },
      "Artifacts": [
        {
          "Uri": "s3://greengrass-agent-test-ipc/agent-test-ipc/1.0.0/agent-test-ipc.zip",
          "Unarchive": "ZIP"
        }
      ]
    }
  ]

"run": "python3 {artifacts:path}/src/main.py"

"run": "python3 {artifacts:decompressedPath}/[YOUR_ARTIFACT_NAME]/src/main.py"

      "Lifecycle": {
        "run": "python3 {artifacts:path}/src/main.py"
      },
      "Artifacts": [
        {
          "Uri": "s3://YOUR_BUCKET/iot-core-publisher/1.0.0/iot-core-publisher.zip",
          "Unarchive": "ZIP"

from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2

try:
    ipc_client = GreengrassCoreIPCClientV2()
    # Use IPC client.
except Exception:
    print('Exception occurred when using IPC.', file=sys.stderr)
    traceback.print_exc()
    exit(1)

    import awsiot.greengrasscoreipc
    from awsiot.greengrasscoreipc.model import (
        PublishToTopicRequest,
        PublishMessage,
        BinaryMessage
    )

- IPCクライアントは 新しいバージョンである V2 を使ってください
	- 参考ドキュメント：https://docs.aws.amazon.com/greengrass/v2/developerguide/interprocess-communication.html#ipc-service-connect
- Pythonのコードは下記のドキュメントに記載のコードを参考にして修正して下さい
	 - https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-iot-core-mqtt.html

- Greengrassデバイスの Python環境には awsiotsdk がインストールされていないので、レシピでインストールする指示を追加してください。実行するのは `python3 -m pip install awsiotsdk` です。

      "Lifecycle": {
        "run": "python3 {artifacts:decompressedPath}/agent-test-ipc/src/main.py"
      },

      ↓　↓　↓（編集後）

      "Lifecycle": {
        "install": "python3 -m pip install awsiotsdk",
        "run": "python3 {artifacts:decompressedPath}/agent-test-ipc/src/main.py"
      },

  DefaultConfiguration:
    outputFile: /tmp/ggc-dummy.csv
    interval: 5