Claude CodeでMQTT検証 — mosquittoでpublishとsubscribeを機械的に確かめる

MQTT連携のコードは、動いたかどうかを目で追いにくい種類の作業です。publishした値がsubscribe側に届いたか、届くまでに何秒かかったか、ブローカーに接続を拒否されていないか。Claude Codeに書かせて「できました」で終わらせると、この3点が未検証のまま残ります。

解決は、ローカルにmosquitto(ブローカー)を立て、同梱のコマンドラインクライアントmosquitto_subとmosquitto_pubを検証コマンドとしてClaude Codeに回させることです。待ち時間に上限を付け、終了コードで合否を決めれば、会話の中で「届いたか」を判定できます。

検証ループの全体像: サブスクライブ側を先に待たせる

Claude Codeに渡す検証は、次の4手順に収まります。

MQTTはpublish/subscribe型なので、受信側が購読を済ませる前に送った(保持されない)メッセージは誰にも届きません。この順序の問題が、検証スクリプトで最初に押さえる点になります。

前提: ブローカーをループバックだけで起動する

mosquittoのコマンドラインに-pでポートを渡すと、そのリスナーは2.0以降ループバック(自分自身のマシン)にだけバインドされます。設定ファイルでリスナーを定義した場合は、-pの指定は無視されます。

検証用に設定ファイルを用意するなら、次の2行で足ります。

listener 1884 127.0.0.1
allow_anonymous true

allow_anonymousを明示する理由は、既定値にあります。リスナーを設定ファイルに書くと、allow_anonymousの既定はfalseになります。リスナーを1つも定義しないときだけtrue(ローカルマシンからの接続のみ)です。上の2行からallow_anonymous trueを抜くと、クライアントは認証なしで接続を拒否されます。

起動は-cで設定ファイルを指し、-vで詳細ログを出すとClaude Codeが接続や購読の状況を読めます。

mosquitto -c ./mosquitto-test.conf -v

mosquittoコマンド自体は別ターミナルか、Claude Codeのバックグラウンド実行で動かしたままにします。ブローカーとクライアントの入手方法は、使っているOSのパッケージ管理に従ってください。

手順1: 往復を1コマンドで確かめる

まず手で動かして、期待する形を見ます。購読側を起動します。

mosquitto_sub -h 127.0.0.1 -p 1884 -t 'sensors/room1/temperature' \
  -C 1 -W 10 -q 1 -F '%t %q %p'

オプションの意味は次のとおりです。

別のターミナルから送ります。

mosquitto_pub -h 127.0.0.1 -p 1884 -t 'sensors/room1/temperature' \
  -q 1 -m '{"celsius":21.5}'

購読側にはsensors/room1/temperature 1 {"celsius":21.5}のような1行が出て終了します。-Fを省くと既定はペイロードだけの出力で、トピックを見たいときは-vか-Fを使います。

手順2: 合否判定をスクリプトにする

手で確かめた流れを、Claude Codeが繰り返し呼べる形にします。次の例は、購読をバックグラウンドで起動し、1秒待ってから送信し、購読側の終了コードと受信内容で判定します。

#!/usr/bin/env bash
# scripts/mqtt-check.sh — 往復の疎通を判定する(例)
set -u
H=127.0.0.1; P=1884
TOPIC="test/$$/temperature"
OUT=$(mktemp)
 
mosquitto_sub -h "$H" -p "$P" -t "$TOPIC" -C 1 -W 5 -q 1 \
  -F '%t %q %p' >"$OUT" &
SUB=$!
sleep 1   # 購読が成立するまでの待ち。短すぎると取りこぼす
 
mosquitto_pub -h "$H" -p "$P" -t "$TOPIC" -q 1 \
  -m '{"celsius":21.5}' || { echo "publish失敗 rc=$?"; exit 2; }
 
wait "$SUB"; RC=$?
if [ "$RC" -eq 0 ] && grep -q '"celsius":21.5' "$OUT"; then
  echo "OK: $(cat "$OUT")"
else
  echo "NG: sub rc=$RC"; cat "$OUT"; exit 1
fi

トピックに$$(プロセスID)を入れているのは、前回の検証が残したメッセージや別のクライアントと衝突させないためです。sleep 1は購読が確立するまでの経験的な待ちで、環境によっては調整が要ります。

終了コードの読み方

mosquitto_subとmosquitto_pubは、ブローカーにMQTTレベルで接続を拒否されると、終了コードにCONNACKの理由コードを返します。それ以外のエラーはlibmosquittoの戻り値です。

-Vを指定しないときのプロトコルは3.1.1です。つまりallow_anonymousを忘れた失敗は、5で現れます。v5で接続すると135になるので、数字を見れば「ネットワークではなく認可の問題」とClaude Codeが切り分けられます。27は表に「待ちメッセージのタイムアウト」として載っているコードです。-Wで上限を付けた購読が時間内に何も受け取らなかったときの判定に使えますが、どの条件で返るかは使うバージョンで一度確かめてください。

手順3: ペイロードがJSONとして正しいかを機械で確かめる

センサーやデバイスのペイロードがJSONの場合、-F '%J'が使えます。%Jは、メッセージのメタデータとタイムスタンプを含むJSONを出力し、ペイロードを引用せずにそのまま埋め込みます。ペイロードが有効なJSONでないと、標準エラーにError: Message payload is not valid JSON on topic <topic>が出ます。

mosquitto_sub -h 127.0.0.1 -p 1884 -t 'sensors/#' -C 1 -W 5 \
  -F '%J' --pretty

実装が壊れたJSON(末尾のカンマ、シングルクォートなど)を送ってしまうバグは、MQTT連携で起きがちです。%Jの標準エラーを検証スクリプトの失敗条件に含めると、型の崩れをpublish側の実装で止められます。ペイロードが文字列やバイナリのときは、%j(引用・エスケープ付き)か%x(16進)を使います。

手順4: retainとQoSを分けて確かめる

「保持メッセージ」を使う実装は、購読の順序問題が出ないぶん、後始末で別の問題が起きます。送信側に-rを付けると、ブローカーが最後の値を保持します。

# 保持メッセージを置く
mosquitto_pub -h 127.0.0.1 -p 1884 -t 'switches/kitchen/status' -r -m 'on'
# 後から購読しても1件受け取れる
mosquitto_sub -h 127.0.0.1 -p 1884 -t 'switches/kitchen/status' -C 1 -W 5 -v
# 検証後に保持を消す(長さ0のペイロードを-rで送る)
mosquitto_pub -h 127.0.0.1 -p 1884 -t 'switches/kitchen/status' -r -n

ゼロ長のメッセージは保持されず、保持済みのメッセージを消す扱いになります。-nは長さ0のメッセージを送るオプションです。消し忘れると次回の検証が前回の値で通ってしまうので、スクリプトの最後に必ず置きます。

受信側で保持メッセージを無視したいときは-R、保持されたものだけを受けたいときは--retained-onlyです。「新規のメッセージだけで合否を決める」検証には、-C 1 -Rの組み合わせが公式の説明にある使い方です。

Claude Code側の設定: 許可範囲を狭くする

検証コマンドを毎回承認するのは手間ですが、mosquitto_pubを広く許可すると、本番のブローカーに誤送信するリスクが出ます。許可ルールにホストを埋め込めば、ループバック以外へは送れなくなります。

{
  "permissions": {
    "allow": [
      "Bash(mosquitto_pub -h 127.0.0.1 *)",
      "Bash(mosquitto_sub -h 127.0.0.1 *)",
      "Bash(bash scripts/mqtt-check.sh)"
    ],
    "deny": [
      "Bash(*--remove-retained*)"
    ]
  }
}

Bashのルールは、末尾の*(直前に空白)を含めて、コマンド文字列に対してワイルドカードで照合されます。Bash(mosquitto_pub -h 127.0.0.1 *)なら、-h 127.0.0.1で始まるコマンドだけが許可対象です。ホストを省略したコマンド(既定はlocalhost)や、別ホストへの送信は許可されず、確認が出ます。

--remove-retainedは、受信した保持メッセージをブローカーから消すオプションです。サンプルにmosquitto_sub -t '#' --remove-retained --retained-onlyという、ブローカー上の保持メッセージを全部消す例があります。共有ブローカーで走ると影響が大きいため、denyに入れています。denyはallowより優先されます。

CLAUDE.mdには、検証の作法を数行で書いておきます。

## MQTT の検証
- ブローカーは `mosquitto -c ./mosquitto-test.conf` で起動済み(127.0.0.1:1884)
- コマンドは必ず `-h 127.0.0.1 -p 1884` から書き始める(許可ルールが前方一致のため)
- 送受信の確認は `bash scripts/mqtt-check.sh` を実行し、OK が出るまで修正する
- retained を使う変更では、検証の最後に `-r -n` で消す

「-hを先頭に置く」と書くのは、許可ルールの前方一致に合わせるためです。この1行がないと、Claude Codeがmosquitto_pub -t ... -h ...と書いて許可に当たらず、承認待ちで止まります。

終わるまで止めさせない

「mqtt-check.shがOKを出すまで終わらない」は、Stop hookで強制できます。仕組みはagent hookでテスト合格まで止めない設定と同じで、検証コマンドをmqtt-check.shに置き換えるだけです。

よくあるつまずき

ほかの検証との使い分け

MQTTの検証は、Webサーバーと違ってHTTPクライアントでは叩けません。Webフレームワークでは、サーバーを立てずにリクエストを直接渡す方法があり、HonoのAPIをapp.requestで検証する手順がその例です。MQTTは、ブローカーという常駐プロセスが間に立つので、この「サーバーを立てない」近道が使えません。代わりにループバックのブローカーを1つ立て、クライアント2つで往復を見ます。

ゲームのようにディスプレイを要求するコードを、ダミードライバで画面なしに検証する考え方はPygameのヘッドレス検証に近く、環境依存を外してコマンドだけで判定させる点が共通しています。

まとめ

MQTT連携の検証で、Claude Codeに渡す材料は3つです。ループバックだけで待ち受けるブローカー、-Cと-Wで終わりが決まるmosquitto_sub、終了コードと受信内容で判定するスクリプト。許可ルールで-h 127.0.0.1を固定し、保持メッセージの一括削除をdenyに入れておくと、共有ブローカーを誤って触る事故は避けられます。

最初に整えるのは、sleep 1の長さを自分の環境で詰めることです。それが決まれば、以後の修正はClaude Codeが検証スクリプトを回しながら進められます。