如何自动化添加Azure IoT Hub自定义端点及路由以接收Device Twin变更通知？

我刚好做过类似的自动化配置，给你整理了AZ CLI、REST API和Python SDK三种方式的具体操作步骤，亲测可用：

1. 创建自定义端点（以Event Hub为例） #

先确保你已经创建好目标Event Hub资源，然后执行以下命令添加IoT Hub的自定义端点：

az iot hub routing-endpoint create \
  --hub-name "你的IoT Hub名称" \
  --endpoint-name "自定义端点名称" \
  --endpoint-type eventhub \
  --endpoint-resource-group "资源组名称" \
  --endpoint-subscription-id "订阅ID" \
  --endpoint-resource-name "目标Event Hub名称"

如果用Service Bus Queue/Topic作为端点，只需把--endpoint-type参数改成servicebusqueue或servicebustopic即可。

2. 创建Device Twin变更通知路由 #

执行以下命令创建路由，指定数据源为设备孪生变更事件：

az iot hub route create \
  --hub-name "你的IoT Hub名称" \
  --name "孪生变更路由名称" \
  --source twinChangeEvents \
  --endpoint-name "上面创建的自定义端点名称" \
  --enabled true

这里的--source twinChangeEvents是核心配置，确保只转发Device Twin的变更通知。

1. 创建自定义端点 #

发送PUT请求到以下URI：

https://management.azure.com/subscriptions/{订阅ID}/resourceGroups/{资源组名称}/providers/Microsoft.Devices/IotHubs/{IoT Hub名称}/routingEndpoints/{端点类型}/{端点名称}?api-version=2023-06-30

请求头需要包含Authorization（Bearer令牌）和Content-Type: application/json，请求体示例（Event Hub端点）：

{
  "properties": {
    "endpointType": "EventHub",
    "resourceGroup": "资源组名称",
    "subscriptionId": "订阅ID",
    "connectionString": "Event Hub的连接字符串",
    "name": "自定义端点名称"
  }
}

2. 创建路由规则 #

发送PUT请求到以下URI：

https://management.azure.com/subscriptions/{订阅ID}/resourceGroups/{资源组名称}/providers/Microsoft.Devices/IotHubs/{IoT Hub名称}/routes/{路由名称}?api-version=2023-06-30

请求体示例：

{
  "properties": {
    "source": "TwinChangeEvents",
    "endpointNames": ["上面创建的自定义端点名称"],
    "isEnabled": true
  }
}

先安装依赖库：

pip install azure-mgmt-iothub azure-identity

然后用以下代码实现配置：

from azure.identity import DefaultAzureCredential
from azure.mgmt.iothub import IotHubClient
from azure.mgmt.iothub.models import RoutingEndpoint, RouteProperties, Route

# 初始化客户端
credential = DefaultAzureCredential()
subscription_id = "你的订阅ID"
client = IotHubClient(credential, subscription_id)

# 1. 创建自定义Event Hub端点
endpoint_name = "自定义端点名称"
resource_group_name = "资源组名称"
iot_hub_name = "你的IoT Hub名称"
event_hub_connection_string = "Event Hub连接字符串"

routing_endpoint = RoutingEndpoint(
    endpoint_type="EventHub",
    connection_string=event_hub_connection_string,
    name=endpoint_name
)

client.routing_endpoints.create_or_update(
    resource_group_name=resource_group_name,
    resource_name=iot_hub_name,
    endpoint_name=endpoint_name,
    routing_endpoint=routing_endpoint
)

# 2. 创建Device Twin变更路由
route_name = "孪生变更路由名称"
route_properties = RouteProperties(
    source="TwinChangeEvents",
    endpoint_names=[endpoint_name],
    is_enabled=True
)

route = Route(properties=route_properties)

client.routes.create_or_update(
    resource_group_name=resource_group_name,
    resource_name=iot_hub_name,
    route_name=route_name,
    route=route
)

如果需要过滤特定的孪生变更（比如仅监听某个属性的更新），还可以在路由中添加condition参数，比如condition="twin.properties.desired.Temperature != null"。

内容的提问来源于stack exchange，提问作者cacsar