鸿蒙智能体开发实战：37.鸿蒙壁纸大师 - 异步消息推送（Webhook）

引言

鸿蒙壁纸大师支持异步消息推送能力：当壁纸生成完成后，系统可以通过 Webhook 主动向用户推送通知消息，即使用户已经关闭了智能体页面，也能在消息中心收到壁纸生成完成的通知。

本文将深入webhook_push_service.py的完整实现，解析 HMAC-SHA256 签名、推送请求构建、与 A2A 协议的集成等核心内容。

上图展示了壁纸生成完成后通过 Webhook 网关向用户推送通知的完整消息链路

一、Webhook 推送流程

1.1 整体流程

壁纸生成完成 ↓ ┌─────────────────────┐ │ build_webhook_payload │ │ - 生成 UUID 标识 │ │ - 构建 markdown 内容 │ │ - 组装 JSON-RPC 格式 │ └─────────────────────┘ ↓ ┌─────────────────────┐ │ generate_signature │ │ - HMAC-SHA256 签名 │ │ - Base64 编码 │ └─────────────────────┘ ↓ ┌─────────────────────┐ │ send_webhook_push │ │ - 设置请求头 │ │ - X-Access-Key │ │ - X-Sign │ │ - X-Ts │ │ - POST 到华为网关 │ └─────────────────────┘ ↓ 用户收到推送通知

1.2 与 SSE 流程的集成

Webhook 推送作为壁纸生成流程的最后一步，在stream_wallpaper_generation()中调用：

# services/wallpaper_service.py# 在图片生成成功的分支中ifimage_resultandimage_result.get("type")=="image":image_url=image_result.get("url")# 发送 Webhook 消息推送try:push_success=awaitsend_webhook_push(image_url=image_url,prompt=enhanced_prompt,agent_login_session_id=agent_login_session_id,log_id=log_id,push_id=push_id# 从 A2A 请求中提取的 push_id)ifpush_success:logger.info("Webhook 推送成功")else:logger.warning("Webhook 推送失败或跳过")exceptExceptionase:logger.error(f"Webhook 推送异常：{e}")

二、HMAC-SHA256 签名算法

2.1 签名原理

Webhook 请求使用 HMAC-SHA256 算法对请求进行签名，确保：

1. 身份验证：华为网关可以验证请求确实来自已注册的 Agent
2. 数据完整性：防止请求在传输过程中被篡改
3. 防重放攻击：结合时间戳确保请求仅在一定时间内有效

安全提示：HMAC-SHA256 签名使用了 Secret Key，请务必妥善保管，切勿泄露或硬编码在代码中。推荐通过环境变量或密钥管理服务注入。

2.2 签名实现

# services/webhook_push_service.pyimporthmacimporthashlibimportbase64importtimedefgenerate_signature(secret_key:str,ts:str)->str:""" 生成 Webhook 请求签名 签名算法：Base64(HMAC-SHA256(secretKey, ts)) Args: secret_key: Security Key（从华为开发者平台获取） ts: 当前时间戳（毫秒级） Returns: 签名结果（Base64 编码字符串） """message=ts.encode('utf-8')secret=secret_key.encode('utf-8')signature=hmac.new(secret,message,hashlib.sha256).digest()returnbase64.b64encode(signature).decode('utf-8')

算法流程：

1. 将时间戳字符串编码为 bytes
2. 将 Secret Key 编码为 bytes
3. 使用 HMAC-SHA256 对时间戳进行签名
4. 将签名结果进行 Base64 编码

2.3 签名的数学表达

[
\text{signature} = \text{Base64}\left(\text{HMAC-SHA256}(\text{secretKey}, \text{ts})\right)
]

其中，ts是毫秒级 Unix 时间戳，如"1713600000123"。

三、Webhook 请求构建

3.1 请求体结构

# services/webhook_push_service.pyimportuuiddefbuild_webhook_payload(api_id:str,push_id:str,push_text:str,artifact_parts:list,agent_login_session_id:Optional[str]=None)->Dict[str,Any]:""" 构建 Webhook 推送消息的请求体 Args: api_id: Webhook API ID（华为开发者平台配置） push_id: 推送通知使用的唯一 ID（从请求中获取） push_text: 通知消息展示内容（如"您的壁纸已生成完成"） artifact_parts: parts 列表（图片 markdown 格式） agent_login_session_id: 用户授权身份 ID（可选） Returns: Webhook 请求体 """outer_id=str(uuid.uuid4())inner_id=str(uuid.uuid4())artifact_id=str(uuid.uuid4())payload={"jsonrpc":"2.0","id":outer_id,"result":{"id":inner_id,"apiId":api_id,"pushId":push_id,"pushText":push_text,"kind":"task","artifacts":[{"artifactId":artifact_id,"parts":artifact_parts}]}}# 如果提供了授权会话 ID，则添加到请求中ifagent_login_session_id:payload["result"]["agentLoginSessionId"]=agent_login_session_idreturnpayload

3.2 请求头配置

# 生成时间戳和签名ts=str(int(time.time()*1000))signature=generate_signature(settings.WEBHOOK_API_SECRET,ts)# 构建请求头headers={"Content-Type":"application/json","Accept":"application/json","x-hag-trace-id":str(uuid.uuid4()),# 请求追踪 ID"X-Access-Key":settings.WEBHOOK_API_KEY,# 访问密钥"X-Sign":signature,# 签名"X-Ts":ts# 时间戳}

3.3 完整推送实现

# services/webhook_push_service.pyasyncdefsend_webhook_push(image_url:str,prompt:str,agent_login_session_id:Optional[str]=None,log_id:Optional[str]=None,push_id:Optional[str]=None)->bool:""" 发送 Webhook 消息推送 Args: image_url: 生成的图片 URL prompt: 生成图片的提示词 agent_login_session_id: 用户授权身份 ID（可选） log_id: 日志 ID push_id: 推送通知使用的唯一 ID（从请求中获取） Returns: 推送是否成功 """# 1. 检查配置是否完整ifnotall([settings.WEBHOOK_API_KEY,settings.WEBHOOK_API_SECRET,settings.WEBHOOK_API_ID]):logger.warning("Webhook 推送配置不完整，跳过推送")returnFalse# 2. 检查 push_idifnotpush_id:push_id=settings.WEBHOOK_PUSH_IDifnotpush_id:logger.warning("push_id 未提供，跳过推送")returnFalsetry:# 3. 生成时间戳和签名ts=str(int(time.time()*1000))signature=generate_signature(settings.WEBHOOK_API_SECRET,ts)# 4. 构建请求头headers={"Content-Type":"application/json","Accept":"application/json","x-hag-trace-id":str(uuid.uuid4()),"X-Access-Key":settings.WEBHOOK_API_KEY,"X-Sign":signature,"X-Ts":ts}# 5. 构建推送数据（使用 markdown 格式展示图片）image_markdown=f"![壁纸]({image_url})"artifact_parts=[{"kind":"text","text":image_markdown}]# 6. 构建请求体payload=build_webhook_payload(api_id=settings.WEBHOOK_API_ID,push_id=push_id,push_text="您的壁纸已生成完成，请查看",artifact_parts=artifact_parts,agent_login_session_id=agent_login_session_id)# 7. 发送请求到华为 Webhook 网关asyncwithhttpx.AsyncClient(timeout=30.0)asclient:response=awaitclient.post("https://hag.cloud.huawei.com/open-ability-agent/v1/agent-webhook",headers=headers,json=payload)ifresponse.status_code==200:logger.info("Webhook 推送成功")returnTrueelse:logger.error(f"Webhook 推送失败：{response.status_code}")returnFalseexceptExceptionase:logger.error(f"Webhook 推送异常：{e}",exc_info=True)returnFalse

四、push_id 的传递链路

4.1 从 A2A 请求中提取 push_id

push_id 是 Webhook 推送的关键标识，它从 A2A 请求的消息中提取：

# services/wallpaper_service.pydefextract_device_info(message_parts:List[Dict])->Dict[str,Any]:""" 从消息 parts 中提取设备信息（包括 push_id） """device_info={"device_type":None,"display_version":None,"market_name":None,"push_id":None# 推送通知的唯一标识}forpartinmessage_parts:ifpart.get("kind")=="data"andpart.get("data"):variables=part.get("data",{}).get("variables",{})system_variables=variables.get("systemVariables",{})ifsystem_variables.get("device_type"):device_info["device_type"]=system_variables.get("device_type")# 提取 push_idifsystem_variables.get("push_id"):device_info["push_id"]=system_variables.get("push_id")returndevice_info

关键概念：push_id是 Webhook 推送的核心标识，由华为推送服务生成并通过 A2A 请求的systemVariables字段传入。没有有效的push_id，推送请求将被跳过。

4.2 完整数据流

华为推送服务 ↓ （生成 push_id） 小艺 → A2A 请求（systemVariables.push_id） ↓ 壁纸大师服务器 ↓ （extract_device_info 提取 push_id） stream_wallpaper_generation() ↓ （壁纸生成完成） send_webhook_push(push_id=push_id) ↓ 华为 Webhook 网关 ↓ （根据 push_id 推送给指定用户） 用户收到通知

4.3 多层 fallback 机制

# send_webhook_push 中的 push_id 获取优先级# 优先级1: 从 A2A 请求中提取的 push_id（最精确）push_id=extract_device_info(message_parts)["push_id"]# 优先级2: 如果请求中没有，使用配置中的默认 push_id（向后兼容）ifnotpush_id:push_id=settings.WEBHOOK_PUSH_ID# 优先级3: 仍然没有，跳过推送ifnotpush_id:logger.warning("push_id 未提供，跳过推送")returnFalse

五、配置管理

5.1 环境变量配置

# .env# 华为 Webhook 推送配置WEBHOOK_API_KEY=your_api_key_hereWEBHOOK_API_SECRET=your_api_secret_hereWEBHOOK_API_ID=your_api_id_hereWEBHOOK_PUSH_ID=your_push_id_here# 可选，推荐从请求中动态获取

5.2 配置检查逻辑

# services/webhook_push_service.py# 检查配置是否完整ifnotall([settings.WEBHOOK_API_KEY,settings.WEBHOOK_API_SECRET,settings.WEBHOOK_API_ID]):logger.warning("Webhook 推送配置不完整，跳过推送")returnFalse

三个配置项缺一不可，任一缺失都会跳过推送（不会导致主流程失败）。

六、与 A2A 协议的完整集成

6.1 集成位置

Webhook 推送在stream_wallpaper_generation()中的集成位置：

# services/wallpaper_service.pyasyncdefstream_wallpaper_generation(user_prompt,context_messages,task_id,log_id=None,session_id=None,device_type=None,agent_login_session_id=None,push_id=None,...):# ... 提示词优化、图片生成 ...# 图片生成成功后发起 Webhook 推送ifimage_resultandimage_result.get("type")=="image":image_url=image_result.get("url")# Webhook 推送（非阻塞，异常不影响主流程）try:push_success=awaitsend_webhook_push(image_url=image_url,prompt=enhanced_prompt,agent_login_session_id=agent_login_session_id,log_id=log_id,push_id=push_id)exceptExceptionase:logger.error(f"Webhook 推送异常：{e}")# ... 继续发送完成状态 ...

6.2 非阻塞设计

Webhook 推送的设计原则是不影响主流程：

七、安全注意事项

7.1 密钥管理

# 密钥从环境变量读取，不硬编码在代码中WEBHOOK_API_KEY=os.getenv("WEBHOOK_API_KEY","")WEBHOOK_API_SECRET=os.getenv("WEBHOOK_API_SECRET","")

7.2 时间戳校验

华为网关收到请求后，会检查X-Ts时间戳是否在合理范围内（通常 5 分钟内），防止重放攻击。

7.3 签名验证

华为网关使用相同的 HMAC-SHA256 算法和 Secret Key 重新计算签名，与X-Sign比对：

# 华为网关侧验证逻辑（示意）expected_signature=generate_signature(secret_key,received_ts)ifreceived_signature!=expected_signature:reject_request()# 签名不匹配，拒绝请求

八、常见问题与解决方案

在实际开发中，Webhook 推送可能会遇到以下常见问题：

排查建议：遇到推送问题时，首先检查日志中的X-Access-Key、X-Sign、X-Ts请求头是否正确，然后使用x-hag-trace-id在华为平台追踪请求链路。

九、总结

本文深入解析了鸿蒙壁纸大师的 Webhook 异步消息推送机制，核心要点：

1. HMAC-SHA256 签名：通过 Base64(HMAC-SHA256(secretKey, ts)) 实现请求认证和数据完整性
2. 三层 fallback：请求中的 push_id → 配置中的 push_id → 跳过推送
3. 非阻塞设计：Webhook 失败不影响壁纸生成主流程
4. 与 A2A 协议集成：通过extract_device_info()从请求中提取push_id和agentLoginSessionId
5. 安全防护：时间戳防重放、签名防篡改、密钥从环境变量读取

如果这篇文章对你有帮助，欢迎点赞👍、收藏⭐、关注🔔，你的支持是我持续创作的动力！

相关资源：

• 华为 Agent Webhook 推送文档
• FastAPI 官方文档
• 火山引擎 - 豆包大模型
• 火山引擎 - API 文档
• A2A 协议规范 - 华为
• Python asyncio 官方文档
• Webhook 安全最佳实践
• HMAC-SHA256 算法说明
• JSON-RPC 2.0 规范
• Python httpx 异步 HTTP 客户端