feat: 新增可选 DWS A2UI 卡片发送与更新（Python）

A2UI.md

@@ -0,0 +1,70 @@
+# 发送 A2UI 卡片
+
+本 SDK 参考 [dingtalk-aicard](https://github.com/DingTalk-Real-AI/dingtalk-aicard/tree/d2c2e7f05592a346e72486f0e41b744c1189f6fa) 的公开 DWS 接入方式，增加可选发送通道。A2UI 消息使用 `version: "v1.0"`；V0.8 是钉钉规范版本，不能替换消息版本。
+
+## 发送身份与准备
+
+`DwsA2UIClient`（Go 为 `DWSA2UIClient`）调用本机 DWS，用其登录账号和 Profile 发送。显式配置后才启用，发送身份与 Channel 的机器人应用 Token 独立。建议指定固定的 `corpId:userId` Profile，创建和更新始终使用同一个 Profile；适配器拒绝同时选择多个 Profile。
+
+先单独安装并登录 [DingTalk Workspace CLI](https://github.com/DingTalk-Real-AI/dingtalk-workspace-cli)，确认所用版本有以下两个命令：
+
+```bash
+dws chat message send-a2ui-card --help
+dws chat message update-a2ui-card --help
+```
+
+默认执行 `dws`，可配置命令路径、固定前缀参数和执行超时（默认 30 秒）。SDK 使用参数数组执行，始终不调用 shell。DWS 接收一个 `--content` 参数，本适配器将其限制为 UTF-8 64 KiB，回执限制为 8 MiB。
+
+新增发送通道可以独立使用，也可以注入已有 Channel 的配置。自定义 A2UI 通道需实现同样的发送、更新接口；这为其他已验证的投递集成保留扩展点。
+
+## 消息与生命周期
+
+消息是非空数组，元素可为 JSON 对象或已经序列化的 JSON 字符串。适配器将其转换为 DWS 要求的字符串数组，保留消息顺序和业务数据。SDK 检查 JSON、`version`、单一操作和非空 `surfaceId`；组件属性、引用、初始数据与资源需按 dingtalk-aicard 校验。
+
+仓库中的 `example/a2ui-card.json` 是完整创建示例，`example/a2ui-update.json` 是同一 Surface 的完成增量。可在 dingtalk-aicard 仓库内设置 Python 环境并验证创建示例：
+
+```bash
+python3 skills/dingtalk-aicard/scripts/setup_env.py
+.aicard-venv/bin/python skills/dingtalk-aicard/scripts/aicard_lint.py /path/to/sdk/example/a2ui-card.json --preflight new-card --format json
+```
+
+组件与事件定义以 [钉钉规范](https://github.com/DingTalk-Real-AI/dingtalk-aicard/tree/d2c2e7f05592a346e72486f0e41b744c1189f6fa/spec) 为准。
+
+单聊必须提供 `openDingTalkId`，群聊提供 `openConversationId`，选择一个目标。不要把机器人 `staffId`、普通 `userId` 或 `senderId` 当作个人开放标识；适配器不会自动转换这些标识。
+
+创建使用 `send-a2ui-card`，状态为 `PROCESSING`。发送结果保留完整 DWS 回执，并提取服务端 `bizId`。缺少它时返回更新警告；不能把请求侧 `bizCardId` 或 `openTaskId` 当作更新标识，也不要自动再创建一张卡片。
+
+更新使用 `update-a2ui-card`，必须传原卡片 `bizId`、非空增量和 `flowStatus`。保持原 `surfaceId`、组件 ID 和用户表单数据，发送完整 A2UI 消息，不发送 JSON token 片段。静态卡片通过状态 `FINISH` 和非空、保留原状态的增量结束。
+
+状态支持 `PROCESSING`、`INPUTTING`、`FINISH`、`EXECUTING`、`ERROR`、`ABORTED`、`TIMEOUT`、`CONFIRMING`、`CONFIRMED`，兼容字符串 `1` 至 `9`。
+
+## 验证与回调
+
+只有回执明确确认接受请求才返回成功；执行错误、非 JSON 输出、无确认标志、dry-run 及失败回执会返回错误。写入超时或进程失败后，服务端可能已经接收请求；SDK 明确报告结果可能未知，并且不会自动重试或降级为 Markdown 消息。请保留回执并核实实际状态。
+
+DWS 命令成功和本地测试不能证明钉钉客户端渲染或用户点击回调。本次适配器负责发送与更新；DWS 身份下的业务回调须通过相应的 `user_card_action_triggered` 事件通道另外接入，不会自动进入 Channel 的机器人 `cardAction` 回调。
+
+参考仓库没有给出可复用的机器人应用 Token A2UI OpenAPI 实现。本适配器实现它已经公开的 DWS 路径；需要机器人原生发送时，应提供已验证的接口契约并实现可注入发送通道。
+
+## SDK 调用
+
+```python
+import json
+import os
+from pathlib import Path
+from dingtalk_channel_sdk import Config, DingTalkChannel, DwsA2UIClient
+
+ch = DingTalkChannel(config=Config(
+    client_id=os.environ["DD_CLIENT_ID"],
+    client_secret=os.environ["DD_CLIENT_SECRET"],
+    a2ui_client=DwsA2UIClient(profile=os.environ["DWS_PROFILE"]),
+))
+messages = json.loads(Path("example/a2ui-card.json").read_text(encoding="utf-8"))
+result = await ch.send_a2ui_card({"open_dingtalk_id": os.environ["DWS_OPEN_DINGTALK_ID"]}, messages)
+if not result.biz_id:
+    raise RuntimeError(result.update_warning)
+delta = json.loads(Path("example/a2ui-update.json").read_text(encoding="utf-8"))
+await ch.update_a2ui_card(result.biz_id, delta, "FINISH")
+```
+
+在异步函数中执行以上调用。群聊使用 `{"conversation_id": "<openConversationId>"}`。独立使用时直接调用 `DwsA2UIClient.send_card` 和 `update_card`，不需要机器人凭据。

README.md

@@ -103,3 +103,8 @@ Live check: `DD_CLIENT_ID=... DD_CLIENT_SECRET=... python example/livecheck.py`
 ## License
 
 MIT
+
+
+## A2UI 卡片
+
+支持通过显式配置的 DWS 通道发送和更新 A2UI 卡片，使用 DWS 登录身份。接口、完整示例、消息校验和回调范围见 [A2UI 接入说明](A2UI.md)。

README.zh-CN.md

@@ -103,3 +103,8 @@ pytest                # 108 个测试
 ## License
 
 MIT
+
+
+## A2UI 卡片
+
+支持通过显式配置的 DWS 通道发送和更新 A2UI 卡片，使用 DWS 登录身份。接口、完整示例、消息校验和回调范围见 [A2UI 接入说明](A2UI.md)。

example/a2ui-card.json

@@ -0,0 +1,41 @@
+[
+  {
+    "version": "v1.0",
+    "createSurface": {
+      "surfaceId": "channel-sdk-example",
+      "catalogId": "https://dingtalk.com/card/a2ui/catalogs/public/catalog.json"
+    }
+  },
+  {
+    "version": "v1.0",
+    "updateDataModel": {
+      "surfaceId": "channel-sdk-example",
+      "path": "/",
+      "value": {
+        "text": "这是 Channel SDK 的 A2UI 示例。"
+      }
+    }
+  },
+  {
+    "version": "v1.0",
+    "updateComponents": {
+      "surfaceId": "channel-sdk-example",
+      "components": [
+        {
+          "id": "root",
+          "component": "Column",
+          "children": [
+            "body"
+          ]
+        },
+        {
+          "id": "body",
+          "component": "Markdown",
+          "content": {
+            "path": "/text"
+          }
+        }
+      ]
+    }
+  }
+]

example/a2ui-update.json

@@ -0,0 +1,10 @@
+[
+  {
+    "version": "v1.0",
+    "updateDataModel": {
+      "surfaceId": "channel-sdk-example",
+      "path": "/text",
+      "value": "示例已完成；这条消息更新原卡片的数据。"
+    }
+  }
+]

example/a2ui.py

@@ -0,0 +1,25 @@
+"""使用 DWS 的指定 Profile 发送示例卡片并完成原卡片，不需要机器人凭据。"""
+import asyncio
+import json
+import os
+from pathlib import Path
+from dingtalk_channel_sdk import DwsA2UIClient
+
+
+async def main():
+    profile, recipient = os.environ.get("DWS_PROFILE"), os.environ.get("DWS_OPEN_DINGTALK_ID")
+    if not profile or not recipient:
+        raise RuntimeError("请设置 DWS_PROFILE 和 DWS_OPEN_DINGTALK_ID")
+    client = DwsA2UIClient(command=[os.environ.get("DWS_BIN", "dws")], profile=profile)
+    folder = Path(__file__).resolve().parent
+    messages = json.loads((folder / "a2ui-card.json").read_text(encoding="utf-8"))
+    result = await client.send_card({"open_dingtalk_id": recipient}, messages)
+    if not result.biz_id:
+        raise RuntimeError(result.update_warning)
+    delta = json.loads((folder / "a2ui-update.json").read_text(encoding="utf-8"))
+    await client.update_card(result.biz_id, delta, "FINISH")
+    print(f"示例卡片已完成，bizId={result.biz_id}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/dingtalk_channel_sdk/__init__.py

@@ -4,6 +4,7 @@
 from .bot_identity import BotIdentity, BotIdentityProvider
 from .card import ApiError as _ApiErrorAlias  # noqa: F401（re-export 由 httpx 提供）
 from .channel import DingTalkChannel
+from .a2ui import A2UIClient, A2UICardResult, DwsA2UIClient, A2UI_FLOW_STATUSES, serialize_a2ui_messages
 from .config import (
     ChatQueueConfig,
     Config,
@@ -29,6 +30,11 @@
 
 __all__ = [
     "DingTalkChannel",
+    "A2UIClient",
+    "A2UICardResult",
+    "DwsA2UIClient",
+    "A2UI_FLOW_STATUSES",
+    "serialize_a2ui_messages",
     "Config",
     "Reply",
     "OapiClient",

src/dingtalk_channel_sdk/a2ui.py

@@ -0,0 +1,173 @@
+"""按 dingtalk-aicard 的公开接入方式，通过显式配置的 DWS 身份发送 A2UI。"""
+from __future__ import annotations
+
+import asyncio
+import json
+import math
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Protocol, Sequence, Union
+
+A2UI_FLOW_STATUSES = (
+    "PROCESSING", "INPUTTING", "FINISH", "EXECUTING", "ERROR",
+    "ABORTED", "TIMEOUT", "CONFIRMING", "CONFIRMED",
+)
+_OPERATIONS = ("createSurface", "updateComponents", "updateDataModel", "deleteSurface")
+_MISSING_ID = "回执未包含可用的 bizId；请保留回执并核实服务端标识，不要自动重发创建请求。"
+Messages = Sequence[Union[str, Dict[str, Any]]]
+
+
+def _reject_json_constant(value: str):
+    raise ValueError("JSON 不允许非有限数值")
+
+
+def _identifier(value: Any, name: str) -> str:
+    if not isinstance(value, str) or not value.strip() or any(ord(c) < 32 or ord(c) == 127 for c in value):
+        raise ValueError(f"{name} 必须是非空标识")
+    return value.strip()
+
+
+def serialize_a2ui_messages(messages: Messages) -> str:
+    """只检查消息信封；组件和完整创建状态使用 dingtalk-aicard 校验。"""
+    if not isinstance(messages, (list, tuple)) or not messages:
+        raise ValueError("A2UI 消息必须是非空数组")
+    strings: List[str] = []
+    for index, message in enumerate(messages):
+        try:
+            encoded = message if isinstance(message, str) else json.dumps(message, ensure_ascii=False, allow_nan=False, separators=(",", ":"))
+            parsed = json.loads(encoded, parse_constant=_reject_json_constant)
+        except (TypeError, ValueError, OverflowError):
+            raise ValueError(f"A2UI 消息 {index} 不是有效 JSON") from None
+        if not isinstance(parsed, dict) or parsed.get("version") != "v1.0":
+            raise ValueError(f"A2UI 消息 {index} 必须是 version=v1.0 的对象")
+        keys = [key for key in _OPERATIONS if key in parsed]
+        if len(keys) != 1 or not isinstance(parsed[keys[0]], dict):
+            raise ValueError(f"A2UI 消息 {index} 必须包含一个操作")
+        _identifier(parsed[keys[0]].get("surfaceId"), "surfaceId")
+        strings.append(encoded)
+    content = json.dumps(strings, ensure_ascii=False, separators=(",", ":"))
+    if len(content.encode("utf-8")) > 65536:
+        raise ValueError("DWS A2UI content 超过 64 KiB")
+    return content
+
+
+def _envelope_chain(receipt: Dict[str, Any]) -> List[Dict[str, Any]]:
+    chain = []
+    value: Any = receipt
+    for _ in range(5):
+        if not isinstance(value, dict):
+            break
+        chain.append(value)
+        if (value.get("success") is False or value.get("ok") is False or value.get("isError") is True or value.get("error")
+                or value.get("dry_run") is True or value.get("dryRun") is True
+                or ("outcome" in value and value["outcome"] not in ("success", "pending"))):
+            raise RuntimeError("DWS 返回失败回执")
+        value = value.get("data") if value.get("data") is not None else value.get("result")
+    if not any(item.get("success") is True or item.get("ok") is True for item in chain):
+        raise RuntimeError("DWS 回执未明确确认接受请求；发送结果可能未知，请核实后再重试")
+    return chain
+
+
+@dataclass(frozen=True)
+class A2UICardResult:
+    biz_id: Optional[str]
+    receipt: Dict[str, Any]
+    update_warning: Optional[str] = None
+
+
+class A2UIClient(Protocol):
+    """可注入发送通道；接收目标使用 open_dingtalk_id 或 conversation_id。"""
+
+    async def send_card(self, target: Dict[str, str], messages: Messages) -> A2UICardResult: ...
+    async def update_card(self, biz_id: str, messages: Messages, flow_status: str) -> Dict[str, Any]: ...
+
+
+class DwsA2UIClient:
+    """可选 DWS 发送通道。DWS 单独安装、登录，profile 决定发送身份。"""
+
+    def __init__(self, command: Sequence[str] = ("dws",), profile: Optional[str] = None, timeout_s: float = 30.0):
+        if isinstance(command, str) or not command or any(not isinstance(part, str) or not part or "\0" in part for part in command):
+            raise ValueError("command 必须是非空字符串数组")
+        if not math.isfinite(timeout_s) or timeout_s <= 0:
+            raise ValueError("timeout_s 必须大于 0")
+        self.command = tuple(command)
+        self.profile = _identifier(profile, "profile") if profile is not None else None
+        if self.profile and "," in self.profile:
+            raise ValueError("A2UI 发送只允许一个 DWS Profile")
+        self.timeout_s = timeout_s
+
+    async def _invoke(self, args: List[str]) -> Dict[str, Any]:
+        argv = list(self.command) + ["chat", "message"] + args + ["--format=json", "--yes"]
+        if self.profile:
+            argv.append(f"--profile={self.profile}")
+        try:
+            process = await asyncio.create_subprocess_exec(*argv, stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
+        except OSError:
+            raise RuntimeError("DWS 无法启动；请检查安装和 command 配置") from None
+
+        async def read_output():
+            chunks = []
+            size = 0
+            while True:
+                chunk = await process.stdout.read(65536)
+                if not chunk:
+                    break
+                size += len(chunk)
+                if size > 8 * 1024 * 1024:
+                    raise RuntimeError("DWS 回执超过 8 MiB；发送结果可能未知")
+                chunks.append(chunk)
+            await process.wait()
+            return b"".join(chunks)
+
+        try:
+            stdout = await asyncio.wait_for(read_output(), timeout=self.timeout_s)
+        except (asyncio.TimeoutError, asyncio.CancelledError) as error:
+            if process.returncode is None:
+                process.kill()
+            await process.wait()
+            if isinstance(error, asyncio.CancelledError):
+                raise
+            raise RuntimeError("DWS 执行超时；发送结果可能未知，请核实后再重试") from None
+        except RuntimeError:
+            if process.returncode is None:
+                process.kill()
+            await process.wait()
+            raise
+        if process.returncode:
+            raise RuntimeError("DWS 执行失败；发送结果可能未知，请核实后再重试")
+        try:
+            receipt = json.loads(stdout)
+        except (ValueError, UnicodeDecodeError):
+            raise RuntimeError("DWS 输出不是 JSON；发送结果可能未知，请保留现场核实") from None
+        if not isinstance(receipt, dict):
+            raise RuntimeError("DWS 回执必须是 JSON 对象")
+        _envelope_chain(receipt)
+        return receipt
+
+    async def send_card(self, target: Dict[str, str], messages: Messages) -> A2UICardResult:
+        if not isinstance(target, dict) or any(key not in ("open_dingtalk_id", "conversation_id") for key in target):
+            raise ValueError("A2UI target 使用 open_dingtalk_id 或 conversation_id，不接受 user_id")
+        dm, group = "open_dingtalk_id" in target, "conversation_id" in target
+        if dm == group:
+            raise ValueError("A2UI target 必须恰好选择一个接收目标")
+        flag = "open-dingtalk-id" if dm else "conversation-id"
+        target_id = _identifier(target["open_dingtalk_id" if dm else "conversation_id"], flag)
+        content = serialize_a2ui_messages(messages)
+        receipt = await self._invoke(["send-a2ui-card", f"--{flag}={target_id}", f"--content={content}"])
+        biz_id = None
+        for value in reversed(_envelope_chain(receipt)):
+            try:
+                biz_id = _identifier(value.get("bizId"), "bizId")
+                break
+            except ValueError:
+                pass
+        return A2UICardResult(biz_id, receipt, None if biz_id else _MISSING_ID)
+
+    async def update_card(self, biz_id: str, messages: Messages, flow_status: str) -> Dict[str, Any]:
+        card_id = _identifier(biz_id, "bizId")
+        status = str(flow_status).strip().upper()
+        if status in tuple(str(i) for i in range(1, 10)):
+            status = A2UI_FLOW_STATUSES[int(status) - 1]
+        if status not in A2UI_FLOW_STATUSES:
+            raise ValueError("不支持的 A2UI flow_status")
+        content = serialize_a2ui_messages(messages)
+        return await self._invoke(["update-a2ui-card", f"--biz-id={card_id}", f"--content={content}", f"--flow-status={status}"])

src/dingtalk_channel_sdk/channel.py

@@ -10,13 +10,14 @@
 import time
 import urllib.request
 from datetime import timedelta
-from typing import Any, Awaitable, Callable, List, Optional
+from typing import Any, Awaitable, Callable, Dict, List, Optional
 
 from .compat import to_thread
 
 from .safety.batching import BatchConfig, BatchedMessage, MessageBatcher
 from .bot_identity import BotIdentity, BotIdentityProvider
 from .card import CardClient
+from .a2ui import A2UICardResult, Messages
 from .config import TOPIC_BOT_MESSAGE, TOPIC_CARD_CALLBACK, TRANSPORT_HTTP, USER_AGENT, Config
 from .safety.dedup import Deduper
 from .emotion import Emotion
@@ -215,6 +216,18 @@ async def send_text(self, target: SendTarget, content: str) -> None:
     async def send_markdown(self, target: SendTarget, title: str, text: str) -> None:
         await self.sender.send_markdown(target, title, text)
 
+    async def send_a2ui_card(self, target: Dict[str, str], messages: Messages) -> A2UICardResult:
+        """通过显式配置的 A2UI 通道发送，返回 biz_id 与完整回执。"""
+        if self.cfg.a2ui_client is None:
+            raise RuntimeError("请先配置 a2ui_client（例如 DwsA2UIClient）")
+        return await self.cfg.a2ui_client.send_card(target, messages)
+
+    async def update_a2ui_card(self, biz_id: str, messages: Messages, flow_status: str) -> Dict[str, Any]:
+        """用服务端 biz_id 更新同一卡片，flow_status 必填。"""
+        if self.cfg.a2ui_client is None:
+            raise RuntimeError("请先配置 a2ui_client（例如 DwsA2UIClient）")
+        return await self.cfg.a2ui_client.update_card(biz_id, messages, flow_status)
+
     async def send_image(self, target: SendTarget, image_url: str) -> None:
         await self.sender.send_image(target, image_url)