Azure IoT Hub

Azure IoT Hub MCP Server

一个用于 Azure IoT Hub 的 Model Context Protocol（MCP）服务，支持设备注册表、设备孪生、直接方法、作业和消息传递。

范围限定于 IoT Hub 的服务端数据平面（设备管理），不包括 Azure Resource Manager。Microsoft 官方的 Azure MCP Server 已涵盖 ARM 层面的“显示我的 IoT Hub 资源”场景；本服务则支持 Azure MCP 尚不具备的实际设备管理功能。

本服务通过 httpx 直接调用 Azure IoT Hub REST Service API，并自行实现 SAS 令牌生成器，而不使用官方的 azure-iot-hub PyPI 包——该包硬性依赖 uamqp，这是一个在现代 Python 版本和平台上无法成功构建的 C 扩展（已确认：它无法安装在 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)	获取设备身份信息（身份验证机制、状态、连接状态）。
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)	列出/读取设备上的模块身份信息。
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)	读取孪生（标签、期望属性/报告属性）。
update_device_twin(device_id, tags, desired_properties, etag)	合并更新标签/期望属性。
replace_device_twin(device_id, tags, desired_properties, etag)	完全替换标签/期望属性（不进行合并）。
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, ...)	发送云到设备消息。使用 AMQP，而非 REST——请参阅下方注意事项。
purge_c2d_message_queue(device_id)	删除设备消息队列中所有待处理的 C2D 消息。

统计信息

工具	描述
get_device_statistics()	设备总数/启用数/禁用数。
get_service_statistics()	服务级统计信息（例如，已连接设备数）。

重要说明：send_c2d_message 使用 AMQP，而非 REST，且没有真实实例测试覆盖

此服务中的其他所有工具都封装了有文档说明的 REST 端点。向设备发送云到设备消息是唯一的例外：Azure IoT Hub 的 REST API 仅提供 C2D 消息的反馈功能（送达回执）；发送消息只能使用 AMQP 协议（通过阅读官方 azure-iot-hub SDK 源码确认——它将请求交给 AMQP 客户端，而非发起 HTTP 调用）。

为在不引入有问题的 uamqp 依赖的情况下支持此功能，send_c2d_message 直接通过 python-qpid-proton 实现 CBS（基于声明的安全）身份验证握手和消息发送（这是标准的 AMQP 1.0 库，已确认能在现代 Python 版本上正常安装）。协议细节（CBS 字段名、$cbs 节点、/messages/devicebound 目标地址、to 属性格式）已根据 Microsoft 自己在 azure-eventhub/azure-servicebus 中的纯 Python AMQP 实现确认，并非凭空猜测。

此工具没有经过真实实例测试。 Azure IoT Hub 没有免费的模拟器（见下文“测试”），而测试套件的 respx HTTP 模拟无法测试 AMQP。因此，测试改为使用本地虚假 AMQP 端点（同样由 qpid-proton 构建）进行，该端点仅实现足以验证线路协议格式正确的 CBS 握手和 /messages/devicebound 接收器，但不会验证 Azure IoT Hub 的真实业务逻辑。在生产环境中依赖此工具前，请先对照真实 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 门户中本 IoT Hub 的“共享访问策略”下找到。
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	否	覆盖 AMQP 连接目标，而不更改根据连接字符串推导出的 REST 主机名（例如，指向本地测试环境）。

本地运行

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 托管服务存在。