引言

鸿蒙壁纸大师支持异步消息推送能力：当壁纸生成完成后，系统可以通过 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

# 在图片生成成功的分支中
if image_result and image_result.get("type") == "image":
    image_url = image_result.get("url")

    # 发送 Webhook 消息推送
    try:
        push_success = await send_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
        )
        if push_success:
            logger.info("Webhook 推送成功")
        else:
            logger.warning("Webhook 推送失败或跳过")
    except Exception as e:
        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.py
import hmac
import hashlib
import base64
import time


def generate_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()
    return base64.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.py
import uuid


def build_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，则添加到请求中
    if agent_login_session_id:
        payload["result"]["agentLoginSessionId"] = agent_login_session_id

    return payload

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                                     # 时间戳
}

请求头	值	说明
X-Access-Key	Webhook API Key	华为开发者平台分配的 Access Key
X-Sign	Base64 HMAC-SHA256	使用 Secret Key 对时间戳签名
X-Ts	毫秒级时间戳	防重放攻击
x-hag-trace-id	UUID	请求追踪 ID，便于排查问题

3.3 完整推送实现

# services/webhook_push_service.py


async def send_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. 检查配置是否完整
    if not all([settings.WEBHOOK_API_KEY,
                settings.WEBHOOK_API_SECRET,
                settings.WEBHOOK_API_ID]):
        logger.warning("Webhook 推送配置不完整，跳过推送")
        return False

    # 2. 检查 push_id
    if not push_id:
        push_id = settings.WEBHOOK_PUSH_ID
    if not push_id:
        logger.warning("push_id 未提供，跳过推送")
        return False

    try:
        # 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 网关
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                "https://hag.cloud.huawei.com/open-ability-agent/v1/agent-webhook",
                headers=headers,
                json=payload
            )

            if response.status_code == 200:
                logger.info("Webhook 推送成功")
                return True
            else:
                logger.error(f"Webhook 推送失败：{response.status_code}")
                return False

    except Exception as e:
        logger.error(f"Webhook 推送异常：{e}", exc_info=True)
        return False

四、push_id 的传递链路

4.1 从 A2A 请求中提取 push_id

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

# services/wallpaper_service.py

def extract_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          # 推送通知的唯一标识
    }

    for part in message_parts:
        if part.get("kind") == "data" and part.get("data"):
            variables = part.get("data", {}).get("variables", {})
            system_variables = variables.get("systemVariables", {})

            if system_variables.get("device_type"):
                device_info["device_type"] = system_variables.get("device_type")
            # 提取 push_id
            if system_variables.get("push_id"):
                device_info["push_id"] = system_variables.get("push_id")

    return device_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（向后兼容）
if not push_id:
    push_id = settings.WEBHOOK_PUSH_ID

# 优先级3: 仍然没有，跳过推送
if not push_id:
    logger.warning("push_id 未提供，跳过推送")
    return False

五、配置管理

5.1 环境变量配置

# .env

# 华为 Webhook 推送配置
WEBHOOK_API_KEY=your_api_key_here
WEBHOOK_API_SECRET=your_api_secret_here
WEBHOOK_API_ID=your_api_id_here
WEBHOOK_PUSH_ID=your_push_id_here    # 可选，推荐从请求中动态获取

5.2 配置检查逻辑

# services/webhook_push_service.py

# 检查配置是否完整
if not all([settings.WEBHOOK_API_KEY,
            settings.WEBHOOK_API_SECRET,
            settings.WEBHOOK_API_ID]):
    logger.warning("Webhook 推送配置不完整，跳过推送")
    return False

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

六、与 A2A 协议的完整集成

6.1 集成位置

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

# services/wallpaper_service.py

async def stream_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 推送
    if image_result and image_result.get("type") == "image":
        image_url = image_result.get("url")

        # Webhook 推送（非阻塞，异常不影响主流程）
        try:
            push_success = await send_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
            )
        except Exception as e:
            logger.error(f"Webhook 推送异常：{e}")

    # ... 继续发送完成状态 ...

6.2 非阻塞设计

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

特性	实现方式	效果
异常隔离	try/except 包裹	Webhook 失败不影响壁纸返回
配置不完整	提前 return false	不阻塞生成流程
推送失败	仅记录日志	用户无感知
独立超时	30 秒超时	不会延长主流程等待

七、安全注意事项

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)
if received_signature != expected_signature:
    reject_request()  # 签名不匹配，拒绝请求

八、常见问题与解决方案

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

问题	原因	解决方案
推送返回 401	API Key 或 Secret 配置错误	检查环境变量是否正确设置
推送返回 403	签名验证失败	确认时间戳为毫秒级，检查 Secret Key
推送返回 404	push_id 无效或已过期	确认 push_id 从 A2A 请求中正确提取
推送超时	网络问题或网关繁忙	设置 30 秒超时，重试机制可选
推送成功但未收到通知	用户未授权或设备离线	检查 agentLoginSessionId 是否有效

排查建议：遇到推送问题时，首先检查日志中的 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 客户端