引言鸿蒙壁纸大师支持异步消息推送能力当壁纸生成完成后系统可以通过 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_urlimage_result.get(url)# 发送 Webhook 消息推送try:push_successawaitsend_webhook_push(image_urlimage_url,promptenhanced_prompt,agent_login_session_idagent_login_session_id,log_idlog_id,push_idpush_id# 从 A2A 请求中提取的 push_id)ifpush_success:logger.info(Webhook 推送成功)else:logger.warning(Webhook 推送失败或跳过)exceptExceptionase:logger.error(fWebhook 推送异常{e})二、HMAC-SHA256 签名算法2.1 签名原理Webhook 请求使用 HMAC-SHA256 算法对请求进行签名确保身份验证华为网关可以验证请求确实来自已注册的 Agent数据完整性防止请求在传输过程中被篡改防重放攻击结合时间戳确保请求仅在一定时间内有效安全提示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 编码字符串 messagets.encode(utf-8)secretsecret_key.encode(utf-8)signaturehmac.new(secret,message,hashlib.sha256).digest()returnbase64.b64encode(signature).decode(utf-8)算法流程将时间戳字符串编码为 bytes将 Secret Key 编码为 bytes使用 HMAC-SHA256 对时间戳进行签名将签名结果进行 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_idstr(uuid.uuid4())inner_idstr(uuid.uuid4())artifact_idstr(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_idreturnpayload3.2 请求头配置# 生成时间戳和签名tsstr(int(time.time()*1000))signaturegenerate_signature(settings.WEBHOOK_API_SECRET,ts)# 构建请求头headers{Content-Type:application/json,Accept:application/json,x-hag-trace-id:str(uuid.uuid4()),# 请求追踪 IDX-Access-Key:settings.WEBHOOK_API_KEY,# 访问密钥X-Sign:signature,# 签名X-Ts:ts# 时间戳}请求头值说明X-Access-KeyWebhook API Key华为开发者平台分配的 Access KeyX-SignBase64 HMAC-SHA256使用 Secret Key 对时间戳签名X-Ts毫秒级时间戳防重放攻击x-hag-trace-idUUID请求追踪 ID便于排查问题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_idsettings.WEBHOOK_PUSH_IDifnotpush_id:logger.warning(push_id 未提供跳过推送)returnFalsetry:# 3. 生成时间戳和签名tsstr(int(time.time()*1000))signaturegenerate_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_markdownfartifact_parts[{kind:text,text:image_markdown}]# 6. 构建请求体payloadbuild_webhook_payload(api_idsettings.WEBHOOK_API_ID,push_idpush_id,push_text您的壁纸已生成完成请查看,artifact_partsartifact_parts,agent_login_session_idagent_login_session_id)# 7. 发送请求到华为 Webhook 网关asyncwithhttpx.AsyncClient(timeout30.0)asclient:responseawaitclient.post(https://hag.cloud.huawei.com/open-ability-agent/v1/agent-webhook,headersheaders,jsonpayload)ifresponse.status_code200:logger.info(Webhook 推送成功)returnTrueelse:logger.error(fWebhook 推送失败{response.status_code})returnFalseexceptExceptionase:logger.error(fWebhook 推送异常{e},exc_infoTrue)returnFalse四、push_id 的传递链路4.1 从 A2A 请求中提取 push_idpush_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)dataandpart.get(data):variablespart.get(data,{}).get(variables,{})system_variablesvariables.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_idpush_id) ↓ 华为 Webhook 网关 ↓ 根据 push_id 推送给指定用户 用户收到通知4.3 多层 fallback 机制# send_webhook_push 中的 push_id 获取优先级# 优先级1: 从 A2A 请求中提取的 push_id最精确push_idextract_device_info(message_parts)[push_id]# 优先级2: 如果请求中没有使用配置中的默认 push_id向后兼容ifnotpush_id:push_idsettings.WEBHOOK_PUSH_ID# 优先级3: 仍然没有跳过推送ifnotpush_id:logger.warning(push_id 未提供跳过推送)returnFalse五、配置管理5.1 环境变量配置# .env# 华为 Webhook 推送配置WEBHOOK_API_KEYyour_api_key_hereWEBHOOK_API_SECRETyour_api_secret_hereWEBHOOK_API_IDyour_api_id_hereWEBHOOK_PUSH_IDyour_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_idNone,session_idNone,device_typeNone,agent_login_session_idNone,push_idNone,...):# ... 提示词优化、图片生成 ...# 图片生成成功后发起 Webhook 推送ifimage_resultandimage_result.get(type)image:image_urlimage_result.get(url)# Webhook 推送非阻塞异常不影响主流程try:push_successawaitsend_webhook_push(image_urlimage_url,promptenhanced_prompt,agent_login_session_idagent_login_session_id,log_idlog_id,push_idpush_id)exceptExceptionase:logger.error(fWebhook 推送异常{e})# ... 继续发送完成状态 ...6.2 非阻塞设计Webhook 推送的设计原则是不影响主流程特性实现方式效果异常隔离try/except包裹Webhook 失败不影响壁纸返回配置不完整提前 return false不阻塞生成流程推送失败仅记录日志用户无感知独立超时30 秒超时不会延长主流程等待七、安全注意事项7.1 密钥管理# 密钥从环境变量读取不硬编码在代码中WEBHOOK_API_KEYos.getenv(WEBHOOK_API_KEY,)WEBHOOK_API_SECRETos.getenv(WEBHOOK_API_SECRET,)7.2 时间戳校验华为网关收到请求后会检查X-Ts时间戳是否在合理范围内通常 5 分钟内防止重放攻击。7.3 签名验证华为网关使用相同的 HMAC-SHA256 算法和 Secret Key 重新计算签名与X-Sign比对# 华为网关侧验证逻辑示意expected_signaturegenerate_signature(secret_key,received_ts)ifreceived_signature!expected_signature:reject_request()# 签名不匹配拒绝请求八、常见问题与解决方案在实际开发中Webhook 推送可能会遇到以下常见问题问题原因解决方案推送返回 401API Key 或 Secret 配置错误检查环境变量是否正确设置推送返回 403签名验证失败确认时间戳为毫秒级检查 Secret Key推送返回 404push_id 无效或已过期确认 push_id 从 A2A 请求中正确提取推送超时网络问题或网关繁忙设置 30 秒超时重试机制可选推送成功但未收到通知用户未授权或设备离线检查 agentLoginSessionId 是否有效排查建议遇到推送问题时首先检查日志中的X-Access-Key、X-Sign、X-Ts请求头是否正确然后使用x-hag-trace-id在华为平台追踪请求链路。九、总结本文深入解析了鸿蒙壁纸大师的 Webhook 异步消息推送机制核心要点HMAC-SHA256 签名通过 Base64(HMAC-SHA256(secretKey, ts)) 实现请求认证和数据完整性三层 fallback请求中的 push_id → 配置中的 push_id → 跳过推送非阻塞设计Webhook 失败不影响壁纸生成主流程与 A2A 协议集成通过extract_device_info()从请求中提取push_id和agentLoginSessionId安全防护时间戳防重放、签名防篡改、密钥从环境变量读取如果这篇文章对你有帮助欢迎点赞、收藏⭐、关注你的支持是我持续创作的动力相关资源华为 Agent Webhook 推送文档FastAPI 官方文档火山引擎 - 豆包大模型火山引擎 - API 文档A2A 协议规范 - 华为Python asyncio 官方文档Webhook 安全最佳实践HMAC-SHA256 算法说明JSON-RPC 2.0 规范Python httpx 异步 HTTP 客户端
鸿蒙智能体开发实战：37.鸿蒙壁纸大师 - 异步消息推送（Webhook）
发布时间：2026/7/13 5:16:19
引言鸿蒙壁纸大师支持异步消息推送能力当壁纸生成完成后系统可以通过 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_urlimage_result.get(url)# 发送 Webhook 消息推送try:push_successawaitsend_webhook_push(image_urlimage_url,promptenhanced_prompt,agent_login_session_idagent_login_session_id,log_idlog_id,push_idpush_id# 从 A2A 请求中提取的 push_id)ifpush_success:logger.info(Webhook 推送成功)else:logger.warning(Webhook 推送失败或跳过)exceptExceptionase:logger.error(fWebhook 推送异常{e})二、HMAC-SHA256 签名算法2.1 签名原理Webhook 请求使用 HMAC-SHA256 算法对请求进行签名确保身份验证华为网关可以验证请求确实来自已注册的 Agent数据完整性防止请求在传输过程中被篡改防重放攻击结合时间戳确保请求仅在一定时间内有效安全提示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 编码字符串 messagets.encode(utf-8)secretsecret_key.encode(utf-8)signaturehmac.new(secret,message,hashlib.sha256).digest()returnbase64.b64encode(signature).decode(utf-8)算法流程将时间戳字符串编码为 bytes将 Secret Key 编码为 bytes使用 HMAC-SHA256 对时间戳进行签名将签名结果进行 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_idstr(uuid.uuid4())inner_idstr(uuid.uuid4())artifact_idstr(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_idreturnpayload3.2 请求头配置# 生成时间戳和签名tsstr(int(time.time()*1000))signaturegenerate_signature(settings.WEBHOOK_API_SECRET,ts)# 构建请求头headers{Content-Type:application/json,Accept:application/json,x-hag-trace-id:str(uuid.uuid4()),# 请求追踪 IDX-Access-Key:settings.WEBHOOK_API_KEY,# 访问密钥X-Sign:signature,# 签名X-Ts:ts# 时间戳}请求头值说明X-Access-KeyWebhook API Key华为开发者平台分配的 Access KeyX-SignBase64 HMAC-SHA256使用 Secret Key 对时间戳签名X-Ts毫秒级时间戳防重放攻击x-hag-trace-idUUID请求追踪 ID便于排查问题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_idsettings.WEBHOOK_PUSH_IDifnotpush_id:logger.warning(push_id 未提供跳过推送)returnFalsetry:# 3. 生成时间戳和签名tsstr(int(time.time()*1000))signaturegenerate_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_markdownfartifact_parts[{kind:text,text:image_markdown}]# 6. 构建请求体payloadbuild_webhook_payload(api_idsettings.WEBHOOK_API_ID,push_idpush_id,push_text您的壁纸已生成完成请查看,artifact_partsartifact_parts,agent_login_session_idagent_login_session_id)# 7. 发送请求到华为 Webhook 网关asyncwithhttpx.AsyncClient(timeout30.0)asclient:responseawaitclient.post(https://hag.cloud.huawei.com/open-ability-agent/v1/agent-webhook,headersheaders,jsonpayload)ifresponse.status_code200:logger.info(Webhook 推送成功)returnTrueelse:logger.error(fWebhook 推送失败{response.status_code})returnFalseexceptExceptionase:logger.error(fWebhook 推送异常{e},exc_infoTrue)returnFalse四、push_id 的传递链路4.1 从 A2A 请求中提取 push_idpush_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)dataandpart.get(data):variablespart.get(data,{}).get(variables,{})system_variablesvariables.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_idpush_id) ↓ 华为 Webhook 网关 ↓ 根据 push_id 推送给指定用户 用户收到通知4.3 多层 fallback 机制# send_webhook_push 中的 push_id 获取优先级# 优先级1: 从 A2A 请求中提取的 push_id最精确push_idextract_device_info(message_parts)[push_id]# 优先级2: 如果请求中没有使用配置中的默认 push_id向后兼容ifnotpush_id:push_idsettings.WEBHOOK_PUSH_ID# 优先级3: 仍然没有跳过推送ifnotpush_id:logger.warning(push_id 未提供跳过推送)returnFalse五、配置管理5.1 环境变量配置# .env# 华为 Webhook 推送配置WEBHOOK_API_KEYyour_api_key_hereWEBHOOK_API_SECRETyour_api_secret_hereWEBHOOK_API_IDyour_api_id_hereWEBHOOK_PUSH_IDyour_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_idNone,session_idNone,device_typeNone,agent_login_session_idNone,push_idNone,...):# ... 提示词优化、图片生成 ...# 图片生成成功后发起 Webhook 推送ifimage_resultandimage_result.get(type)image:image_urlimage_result.get(url)# Webhook 推送非阻塞异常不影响主流程try:push_successawaitsend_webhook_push(image_urlimage_url,promptenhanced_prompt,agent_login_session_idagent_login_session_id,log_idlog_id,push_idpush_id)exceptExceptionase:logger.error(fWebhook 推送异常{e})# ... 继续发送完成状态 ...6.2 非阻塞设计Webhook 推送的设计原则是不影响主流程特性实现方式效果异常隔离try/except包裹Webhook 失败不影响壁纸返回配置不完整提前 return false不阻塞生成流程推送失败仅记录日志用户无感知独立超时30 秒超时不会延长主流程等待七、安全注意事项7.1 密钥管理# 密钥从环境变量读取不硬编码在代码中WEBHOOK_API_KEYos.getenv(WEBHOOK_API_KEY,)WEBHOOK_API_SECRETos.getenv(WEBHOOK_API_SECRET,)7.2 时间戳校验华为网关收到请求后会检查X-Ts时间戳是否在合理范围内通常 5 分钟内防止重放攻击。7.3 签名验证华为网关使用相同的 HMAC-SHA256 算法和 Secret Key 重新计算签名与X-Sign比对# 华为网关侧验证逻辑示意expected_signaturegenerate_signature(secret_key,received_ts)ifreceived_signature!expected_signature:reject_request()# 签名不匹配拒绝请求八、常见问题与解决方案在实际开发中Webhook 推送可能会遇到以下常见问题问题原因解决方案推送返回 401API Key 或 Secret 配置错误检查环境变量是否正确设置推送返回 403签名验证失败确认时间戳为毫秒级检查 Secret Key推送返回 404push_id 无效或已过期确认 push_id 从 A2A 请求中正确提取推送超时网络问题或网关繁忙设置 30 秒超时重试机制可选推送成功但未收到通知用户未授权或设备离线检查 agentLoginSessionId 是否有效排查建议遇到推送问题时首先检查日志中的X-Access-Key、X-Sign、X-Ts请求头是否正确然后使用x-hag-trace-id在华为平台追踪请求链路。九、总结本文深入解析了鸿蒙壁纸大师的 Webhook 异步消息推送机制核心要点HMAC-SHA256 签名通过 Base64(HMAC-SHA256(secretKey, ts)) 实现请求认证和数据完整性三层 fallback请求中的 push_id → 配置中的 push_id → 跳过推送非阻塞设计Webhook 失败不影响壁纸生成主流程与 A2A 协议集成通过extract_device_info()从请求中提取push_id和agentLoginSessionId安全防护时间戳防重放、签名防篡改、密钥从环境变量读取如果这篇文章对你有帮助欢迎点赞、收藏⭐、关注你的支持是我持续创作的动力相关资源华为 Agent Webhook 推送文档FastAPI 官方文档火山引擎 - 豆包大模型火山引擎 - API 文档A2A 协议规范 - 华为Python asyncio 官方文档Webhook 安全最佳实践HMAC-SHA256 算法说明JSON-RPC 2.0 规范Python httpx 异步 HTTP 客户端
