如何在Azure Event Grid Python v2输出绑定中设置CloudEventV10 schema

问题原因 #

你的Event Grid主题配置为接收CloudEventV10格式，但当前代码使用的func.EventGridOutputEvent生成的是Event Grid Schema格式事件，且输出绑定默认会将事件包装为数组，与主题期望的单个CloudEvent对象格式不匹配，导致报错。

解决方案 #

需修改输出绑定配置和事件构造逻辑，适配CloudEventV10格式：

1. 指定输出Schema类型 #

在@app.event_grid_output装饰器中添加schema="CloudEventV10"参数，明确告知输出绑定使用CloudEvent V1.0格式序列化事件。

2. 替换事件构造类 #

使用func.CloudEvent替代func.EventGridOutputEvent，CloudEvent的字段与Event Grid Schema对应关系如下：

• source对应原subject
• type对应原event_type
• specversion固定为1.0（CloudEvent V1.0版本标识）

3. 调整输出参数类型 #

将outputEvent: func.Out[func.EventGridOutputEvent]改为outputEvent: func.Out[func.CloudEvent]，匹配CloudEvent的输出类型。

修改后的完整代码 #

import logging
import azure.functions as func
import datetime

@app.function_name(name="eventgrid_output")
@app.route(route="eventgrid_output")
@app.event_grid_output(
    arg_name="outputEvent",
    topic_endpoint_uri="MyEventGridTopicUriSetting",
    topic_key_setting="MyEventGridTopicKeySetting",
    schema="CloudEventV10"  # 指定输出为CloudEventV10格式
)
def eventgrid_output(eventGridEvent: func.EventGridEvent, 
         outputEvent: func.Out[func.CloudEvent]) -> None:

    logging.info(f"eventGridEvent: {eventGridEvent}")

    # 构造CloudEventV10格式的事件
    outputEvent.set(
        func.CloudEvent(
            id="test-id",
            source="test-subject",
            type="test-event-1",
            time=datetime.datetime.utcnow(),
            data={"tag1": "value1", "tag2": "value2"},
            specversion="1.0"
        )
    )

关键说明 #

• schema="CloudEventV10"参数确保输出事件以单个CloudEvent对象形式发送，而非数组，完全匹配主题的接收格式要求。
• func.CloudEvent是Azure Functions Python SDK中专门适配CloudEvent规范的类，字段结构完全符合CloudEvent V1.0标准。

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