Azure Function发Service Bus消息如何指定content-type为application/json

Python 版 Azure Function 的 Service Bus 输出绑定在直接传入字符串类型消息时，运行时会默认将消息的 content-type 标记为 text/plain，仅通过 msg.set() 传入字符串无法修改该属性，需要通过构造原生 Service Bus 消息对象的方式自定义属性，具体修改步骤如下：

• 导入 azure.servicebus.ServiceBusMessage 类，这是 Service Bus 原生的消息结构，支持自定义所有消息属性
• 将函数签名中输出绑定的类型从 func.Out[str] 修改为 func.Out[ServiceBusMessage]
• 构造消息实例时显式传入 content_type="application/json" 参数，再将消息实例传给 msg.set()
• 修复你现有代码里的双重序列化问题：get_json_content_from_csv 已经返回了 json.dumps() 处理后的合法 JSON 字符串，不需要再对结果做一次 json.dumps()，否则消息内容会出现多余转义符。

修改后的完整 __init__.py 代码如下：

import io
import csv
import json
from azure.functions import EventGridEvent, Out
from azure.storage.blob import BlobClient
from azure.identity import DefaultAzureCredential
from azure.servicebus import ServiceBusMessage


def get_json_content_from_csv(csv_content: str) -> str:
    reader = csv.DictReader(io.StringIO(csv_content))
    json_content = json.dumps(list(reader))
    return json_content


def main(event: EventGridEvent, msg: Out[ServiceBusMessage]):
    data = event.get_json()
    url = data["url"]
    input_blob = BlobClient.from_blob_url(url, DefaultAzureCredential())
    csv_content = input_blob.download_blob(encoding='UTF-8').readall()
    json_content = get_json_content_from_csv(csv_content)
    # 构造消息时指定content_type
    sb_message = ServiceBusMessage(
        json_content,
        content_type="application/json"
    )
    msg.set(sb_message)

你的 function.json 配置不需要做任何修改，现有绑定规则可以直接适配上述代码。修改后发送的消息，在 Service Bus 主题/订阅侧查看属性，即可看到 content-type 已正确设置为 application/json，接收端可直接按 JSON 格式解析内容。

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