Azure IoT Hub

Azure IoT Hub MCP サーバー

Azure IoT Hub — デバイスレジストリ、デバイスツイン、ダイレクトメソッド、ジョブ、メッセージング向けの Model Context Protocol（MCP）サーバーです。

対象は IoT Hub のサービス側データプレーン（デバイス管理）であり、Azure Resource Managerではありません。Microsoft 公式の Azure MCP Server は、すでに ARM レベルの「自分の IoT Hub リソースを表示する」ケースに対応しています。このサーバーは、Azure MCP が対応していない実際のデバイス管理機能を提供します。

公式の azure-iot-hub PyPI パッケージではなく、httpx を使って Azure IoT Hub の REST Service API を直接呼び出し、独自に実装した SAS トークン生成機能を使用します。このパッケージは uamqp に強く依存しており、この C 拡張機能は最新の Python／プラットフォームではビルドに失敗します（確認済み：Python 3.12／arm64 にはインストールできません）。FastMCP で構築されています。

ツール

デバイスレジストリ

ツール	説明
list_devices(max_count)	デバイスを一覧表示します。絞り込みには query_devices を推奨します。
query_devices(query, max_item_count, continuation_token)	IoT Hub の SQL に似たクエリで、デバイス／ツインをページネーション付きで検索します。
get_device(device_id)	デバイスの ID 情報（認証メカニズム、状態、接続状態）を取得します。
create_device(device_id, auth_type, ..., overwrite=False)	デバイスを登録します。
update_device(device_id, status, status_reason, etag)	デバイスの有効／無効状態を更新します。
delete_device(device_id, confirm=False, etag)	デバイスを削除します。取り消しできません。

モジュール

ツール	説明
list_modules(device_id) / get_module(device_id, module_id)	デバイス上のモジュール ID を一覧表示／読み取ります。
create_module(device_id, module_id, ..., overwrite=False)	モジュールを登録します。
delete_module(device_id, module_id, confirm=False)	モジュールを削除します。取り消しできません。

デバイスとモジュールのツイン

ツール	説明
get_device_twin(device_id) / get_module_twin(device_id, module_id)	ツイン（タグ、desired／reported プロパティ）を読み取ります。
update_device_twin(device_id, tags, desired_properties, etag)	タグ／desired プロパティをマージ更新します。
replace_device_twin(device_id, tags, desired_properties, etag)	タグ／desired プロパティを完全に置き換えます（マージではありません）。
update_module_twin(device_id, module_id, tags, desired_properties, etag)	モジュールツインをマージ更新します。

ダイレクトメソッド

ツール	説明
invoke_device_method(device_id, method_name, payload, ...)	デバイスのダイレクトメソッドを呼び出します（デバイスが応答するか、タイムアウトするまで待機します）。
invoke_module_method(device_id, module_id, method_name, payload, ...)	同じ処理をモジュールに対して実行します。

ジョブ（大規模なデバイス群に対する操作）

ツール	説明
create_scheduled_job(job_id, job_type, query_condition, ...)	クエリに一致する多数のデバイスに対して、ツイン更新またはダイレクトメソッドをスケジュールします。
get_scheduled_job(job_id) / query_scheduled_jobs(job_type, job_status)	ジョブの状態／結果を読み取ります。
cancel_scheduled_job(job_id, confirm=False)	ジョブをキャンセルします。

メッセージング

ツール	説明
send_c2d_message(device_id, payload, ...)	クラウドからデバイスへのメッセージを送信します。REST ではなく AMQP を使用します。下記の注意事項を参照してください。
purge_c2d_message_queue(device_id)	デバイス宛てにキューイングされた保留中の C2D メッセージをすべて削除します。

統計情報

ツール	説明
get_device_statistics()	デバイスの合計数／有効数／無効数を取得します。
get_service_statistics()	サービスレベルの統計情報（接続中のデバイス数など）を取得します。

重要な注意点：send_c2d_message は REST ではなく AMQP であり、実インスタンスでのテストカバレッジはありません

このサーバーの他のすべてのツールは、ドキュメント化された REST エンドポイントをラップしています。クラウドからデバイスへのメッセージの送信だけは例外です。Azure IoT Hub の REST API が公開しているのは、C2D メッセージングのフィードバック側（配信確認）のみであり、送信には AMQP プロトコルしか使えません（公式の azure-iot-hub SDK のソースを確認したところ、HTTP 呼び出しではなく AMQP クライアントに処理を委譲していました）。

壊れた uamqp 依存関係を取り込まずにこれをサポートするため、send_c2d_message は python-qpid-proton（標準的な AMQP 1.0 ライブラリ。最新の Python に問題なくインストールできることを確認済み）を使って、CBS（Claims-Based Security）認証ハンドシェイクとメッセージ送信を直接実装しています。プロトコルの詳細（CBS のフィールド名、$cbs ノード、/messages/devicebound の宛先アドレス、to プロパティの形式）は、推測ではなく、Microsoft 独自の純粋な Python による AMQP 実装である azure-eventhub／azure-servicebus で確認しました。

このツールは実インスタンスでのテスト対象ではありません。 無料で使える Azure IoT Hub エミュレーターはなく（下記のテストを参照）、テストスイートの respx による HTTP モックでは AMQP を一切検証できません。そのため、テストでは代わりにローカルの偽 AMQP エンドポイント（こちらも qpid-proton で構築）を相手に実行します。このエンドポイントは、ワイヤプロトコルが適切な形式になっていることを確認できるよう、CBS ハンドシェイクと /messages/devicebound レシーバーを最低限実装していますが、本物の Azure IoT Hub のビジネスロジックを検証するものではありません。本番環境でこのツールに依存する前に、実際のハブで動作を確認してください。

安全性に関する規約

このリポジトリの他の部分と同様に、create_* はデフォルトで overwrite=False です（ベストエフォートの説明取得後に作成するチェックを行います。IoT Hub の If-Match ヘッダーは作成専用のセマンティクスではなく、更新時の楽観的同時実行制御に使われるため、Ditto の If-None-Match のように完全に競合を防げるわけではありません）。すべての delete_*／cancel_* では confirm=True が必要です。

設定

環境変数	必須	説明
AZURE_IOT_HUB_CONNECTION_STRING	はい	service（または iothubowner）の権限を持つ IoT Hub 接続文字列です。例：HostName=my-hub.azure-devices.net;SharedAccessKeyName=service;SharedAccessKey=...。Azure ポータルで、ハブの「共有アクセス ポリシー」から確認できます。
AZURE_IOT_HUB_AMQP_PORT	いいえ	デフォルトは 5671 です。send_c2d_message にのみ関係します。
AZURE_IOT_HUB_AMQP_USE_TLS	いいえ	true／false。デフォルトは true です。ローカル／テスト用エンドポイントの場合に限り、false に設定してください。
AZURE_IOT_HUB_AMQP_HOSTNAME_OVERRIDE	いいえ	接続文字列から導出される REST ホスト名は変更せずに、AMQP の接続先を上書きします（例：ローカルのテスト用フィクスチャ）。

ローカルでの実行

uv sync
export AZURE_IOT_HUB_CONNECTION_STRING="HostName=my-hub.azure-devices.net;SharedAccessKeyName=service;SharedAccessKey=..."
uv run azureiothubmcpserver.py

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

テスト

このリポジトリの設計原則では、モックではなく実際のインスタンスを使ったテストを推奨しています。Azure IoT Hub には公式のローカルエミュレーターがなく、moto に相当するものもありません。REST エンドポイントは、respx（httpx 専用のモックライブラリ。Azure の実際の応答形式に即したデータを使用）でテストします。send_c2d_message は、ローカルの偽 AMQP エンドポイントを相手にテストします（上記の注意点を参照）。

uv run pytest -v

ライセンスに関する注記

ローカルで実行できるブローカー／サービスがないため、ここにはデプロイ用アーティファクトを同梱していません（eclipse-ditto／mqtt とは異なります）。Azure IoT Hub は Azure 上でホストされるサービスとしてのみ提供されています。