MQTT

MQTT MCP サーバー

MQTT 用の Model Context Protocol（MCP）サーバーです。MQTT は標準的な IoT パブリッシュ／サブスクライブ型メッセージングプロトコルです。あらゆる MQTT ブローカー（Eclipse Mosquitto、EMQX、HiveMQ、AWS IoT Core、Azure IoT Hub の MQTT エンドポイントなど）で動作します。使用するのは標準の MQTT 5.0 ワイヤープロトコルのみで、ブローカー固有の管理 API は使用しません。

FastMCP と paho-mqtt を使用して構築されています。

ツール

ツール	説明
get_connection_status()	このコネクターのブローカー接続状態を報告します。
publish_message(topic, payload, qos, retain, ...)	メッセージをパブリッシュします。MQTT5 のプロパティ（コンテンツタイプ、メッセージの有効期限、応答トピック、相関データ、ユーザープロパティ）をサポートし、必要に応じてブローカーが配信を確認するまでブロックします（QoS 1/2）。
start_subscription(topic_filter, qos, ...)	トピックフィルターに一致するメッセージの受信を開始します（+/# ワイルドカードをサポート）。session_id を返します。
get_messages(session_id, max_messages, timeout_seconds)	サブスクリプションセッションのバッファーにあるメッセージを取り出します。メッセージが届くまで短時間待機することもできます。
stop_subscription(session_id)	セッションを停止してサブスクライブを解除します（同じフィルターを必要とする他のセッションがない場合）。
list_subscriptions()	アクティブなサブスクリプションセッションをすべて一覧表示します。
discover_topics(topic_filter, duration_seconds, max_topics)	保持メッセージを使って「トピック一覧」操作を近似します。MQTT にはネイティブのトピック一覧取得操作がありません。

サブスクリプションがセッションベースである理由

MQTT は継続的なパブリッシュ／サブスクライブストリームであり、リクエスト／レスポンス API ではありません。MCP ツールの呼び出しを 1 回行うだけでメッセージを「永遠に待つ」ことはできないため、サブスクライブは 2 段階で行います。start_subscription で対象を登録し、バックグラウンドでメッセージをバッファーに保存します（ブローカー接続は 1 つを共有し、トピックフィルターに基づいてプロセス内で振り分けます）。続いて get_messages で、必要な回数だけツールを呼び出してバッファーからメッセージを取り出します。このとき、新しいデータを一定時間だけ待つこともできます。完了したら stop_subscription を呼び出してください。

設定

環境変数	必須	説明
MQTT_HOST	はい	ブローカーのホスト名または IP。
MQTT_PORT	いいえ	デフォルトは 1883 です（TLS の場合は 8883。明示的に設定してください）。
MQTT_USERNAME / MQTT_PASSWORD	いいえ	ブローカーの認証情報。
MQTT_CLIENT_ID	いいえ	デフォルトではブローカーによって割り当てられる ID を使用します。
MQTT_KEEPALIVE	いいえ	秒数。デフォルトは 60 です。
MQTT_CONNECT_TIMEOUT	いいえ	初回接続を待つ秒数。デフォルトは 10 です。
MQTT_TRANSPORT	いいえ	tcp（デフォルト）または websockets。
MQTT_WS_PATH	いいえ	WebSocket トランスポートのパス。デフォルトは /mqtt です。
MQTT_USE_TLS	いいえ	true/false。デフォルトは false です。
MQTT_TLS_CA_CERTS	いいえ	CA バンドルへのパス。省略すると、システムの信頼ストアを使用します。
MQTT_TLS_CLIENT_CERT / MQTT_TLS_CLIENT_KEY	いいえ	相互 TLS 用です。
MQTT_TLS_INSECURE	いいえ	ホスト名の検証をスキップします。自己署名テスト証明書の場合にのみ使用してください。
MQTT_WILL_TOPIC / MQTT_WILL_PAYLOAD / MQTT_WILL_QOS / MQTT_WILL_RETAIN	いいえ	最後の意思表示（Last Will and Testament）。接続時に一度設定します。

ローカルでの実行

uv sync
export MQTT_HOST=localhost
export MQTT_PORT=1883
export MQTT_USERNAME=mqtt
export MQTT_PASSWORD=mqtt
uv run mqttmcpserver.py

すぐに使えるクライアント設定（stdio および Docker）については、mcp.json を参照してください。

実際のブローカーを使ったテスト

このリポジトリの設計原則では、モックではなく実際のインスタンスを使ったテストを求めています。docker-compose.yml は、パスワード認証を有効にした Eclipse Mosquitto（2.x、MQTT5 対応）と、TLS/mTLS のコードパスをテストするための自己署名証明書を使用する TLS リスナーを起動します。

docker compose up -d       # start Mosquitto on localhost:1883 (plain, user: mqtt / pass: mqtt) and :8883 (TLS)
uv run pytest -v           # tests/conftest.py also starts/stops the stack automatically
docker compose down        # stop it manually if you started it yourself

ライセンスに関する注記

docker-compose.yml と docker/ 内のファイルは、ローカルテスト専用にオープンソースの Eclipse Mosquitto ブローカー（EPL-2.0）を設定するものであり、MCP サーバー自体の一部ではありません。