mcpField
M
mqtt

mqtt

github.com/alphaonedev開発52 回インストール
リポジトリ

mqtt について

AIがIoT通信でMQTTのパブリッシュ／サブスクライブを扱うためのスキルです。テレメトリーの送信やデバイス更新の受信に加え、接続の安全性と信頼性に関する設定を扱います。

主な機能

  1. 01QoSレベルを指定したトピック送信
  2. 02非同期の購読とメッセージ処理
  3. 03保持メッセージとワイルドカードトピック
  4. 04TLSとユーザー名・パスワード認証

活用シーン

  1. 01IoTダッシュボードへのセンサー値送信
  2. 02デバイス制御コマンドの受信

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/alphaonedev/openclaw-graph --skill mqtt

インストール後、Claude Code で/mqttを入力して読み込まれたか確認します。

詳細

mqtt

目的

このスキルを使うと、AI は軽量なパブリッシュ／サブスクライブプロトコルである MQTT を使用して、効率的な IoT 通信を行えます。テレメトリーデータのパブリッシュやデバイスの更新情報のサブスクライブに利用し、制約のある環境でもリアルタイムのやり取りを可能にします。

使用する場面

IoT デバイスからのセンサーデータの送信、リモートデバイスの制御、テレメトリーの監視など、低帯域幅でリアルタイムのメッセージングが必要な場面で使用します。ファイル転送など、高スループットが求められる用途では HTTP のほうが適しているため、使用を避けてください。

主な機能

  • 設定可能な Quality of Service（QoS）レベル（0、1、2）で、トピックにメッセージをパブリッシュします。
  • トピックをサブスクライブし、受信メッセージを非同期で処理します。
  • 保持メッセージ、遺言メッセージ、ワイルドカードサブスクリプション（例：「#」ですべてのサブトピックを対象）をサポートします。
  • TLS と、ユーザー名／パスワードなどの認証方式を使って安全に接続します。

使用パターン

MQTT を使用するには、まずクライアントライブラリを使ってブローカーに接続します。認証情報を使って接続を確立してから、パブリッシュまたはサブスクライブします。たとえば、ループ内で接続、サブスクライブ、メッセージの処理、切断を行います。ブローカーアドレスには $MQTT_BROKER_URL のような環境変数を使い、機密データを扱います。不安定なネットワークでは、必ず再接続を処理してください。

主なコマンド／API

Python での操作には Paho MQTT ライブラリを使用します。次のようにインポートします。

import paho.mqtt.client as mqtt

ブローカーに接続するには：

  • コマンド：client.connect(broker_address, port=1883)
  • 認証を使う例：client.username_pw_set(username='$MQTT_USERNAME', password='$MQTT_PASSWORD')

メッセージをパブリッシュするには：

  • API：client.publish(topic, payload, qos=1)
  • 例：client.publish('sensors/temp', '22.5', qos=1)

サブスクライブするには：

  • API：client.subscribe(topic, qos=1)
  • 例：client.subscribe('commands/device1', qos=1)

設定形式：設定には JSON ファイルを使用します。例：{"broker": "$MQTT_BROKER_URL", "topic": "sensors/temp"}。mosquitto_pub などの CLI ツールは、subprocess を使って呼び出せます：subprocess.run(['mosquitto_pub', '-h', '$MQTT_BROKER_URL', '-t', 'sensors/temp', '-m', '22.5'])。

統合に関する注意

MQTT を AI エージェントのコードでラップして統合します。トークンベース認証用のキーには、export $MQTT_API_KEY のように環境変数を設定します。複数のサービスを使う構成では、HiveMQ や Mosquitto などのブローカーを使用し、client.connect('$MQTT_BROKER_URL') で接続します。セキュアなエンドポイントでは TLS を有効にしてください（例：client.tls_set()）。データベースなどのほかのスキルと組み合わせる場合は、処理が成功したときにデータをパブリッシュします。たとえば、DB への挿入後に client.publish('db/updates', json.dumps(data)) を実行します。

エラー処理

よくあるエラーは明示的に処理してください。接続の問題については、ConnectionRefusedError を捕捉し、指数バックオフ（例：time.sleep(5)）で再試行します。パブリッシュに失敗した場合は、戻り値を確認します（例：client.publish() が >0 を返した場合にログを記録して再試行します）。サブスクリプションエラーは、無効なトピックが原因である可能性があります。if '/' not in topic: raise ValueError('Invalid topic') のように検証してください。on_message ハンドラーで if error: print('Error: ', error) のように、コールバックを使ってメッセージエラーを処理します。必ず finally ブロックで client.disconnect() を呼び出して後始末してください。

具体的な使用例

例1：センサーデータをパブリッシュする 温度の測定値を MQTT トピックにパブリッシュするには、まず環境変数を設定します：export MQTT_BROKER_URL='broker.example.com'。次に、コードで以下を実行します：

client = mqtt.Client()
client.connect(os.environ['MQTT_BROKER_URL'])
client.publish('sensors/temperature', '25.0', qos=1)
client.disconnect()

これにより、IoT ダッシュボードの更新用メッセージがブローカーに送信されます。

例2：デバイスコマンドをサブスクライブする コマンドをサブスクライブして応答するには、次のようにします：

def on_message(client, userdata, message):
    print(f"Received: {message.payload.decode()}")
client = mqtt.Client()
client.on_message = on_message
client.connect(os.environ['MQTT_BROKER_URL'])
client.subscribe('commands/device1', qos=1)
client.loop_forever()

これにより、クライアントはコマンドを受信し続け、IoT デバイスを制御できます。

グラフの関係

  • 関連するクラスター：iot
  • 関連するタグ：iot, mqtt
  • 接続：データ処理やデバイス管理など、ほかの iot スキルと統合できます。

コメント

Markdown 対応