A2A Agent Card 与任务模型

mkdir l6-a2a
cd l6-a2a

"""按A2A 1.0字段编写的本地对象与业务状态检查。"""
from copy import deepcopy
from urllib.parse import urlparse

CARD = {
    "name": "退款资料助手", "description": "读取退款规则并整理依据",
    "version": "1.0.0",
    "supportedInterfaces": [{
        "url": "https://refund.example/a2a", "protocolBinding": "JSONRPC",
        "protocolVersion": "1.0",
    }],
    "capabilities": {"streaming": False},
    "defaultInputModes": ["text/plain"], "defaultOutputModes": ["text/plain"],
    "skills": [{"id": "refund.read", "name": "退款规则读取",
                "description": "按订单查询可用规则", "tags": ["refund"]}],
}
MESSAGE = {"messageId": "M-17", "role": "ROLE_USER",
           "parts": [{"text": "查询O-17的退款规则"}]}
STATES = {"TASK_STATE_" + value for value in (
    "UNSPECIFIED", "SUBMITTED", "WORKING", "COMPLETED", "FAILED",
    "CANCELED", "INPUT_REQUIRED", "REJECTED", "AUTH_REQUIRED")}
TERMINAL = {"TASK_STATE_" + value for value in ("COMPLETED", "FAILED", "CANCELED", "REJECTED")}
NEXT = {
    "TASK_STATE_SUBMITTED": {"TASK_STATE_WORKING", "TASK_STATE_REJECTED"},
    "TASK_STATE_WORKING": {"TASK_STATE_INPUT_REQUIRED", "TASK_STATE_AUTH_REQUIRED"} | TERMINAL,
    "TASK_STATE_INPUT_REQUIRED": {"TASK_STATE_WORKING", "TASK_STATE_CANCELED"},
    "TASK_STATE_AUTH_REQUIRED": {"TASK_STATE_WORKING", "TASK_STATE_CANCELED"},
}

def text(value):
    return isinstance(value, str) and bool(value.strip())

def strings(value):
    return isinstance(value, list) and bool(value) and all(text(item) for item in value)

def validate_card(card):
    # 检查当前例子所用的字段和JSONRPC 1.0接口配置。
    for key in ("name", "description", "version"):
        if not text(card.get(key)):
            raise ValueError("AgentCard缺少" + key)
    if not isinstance(card.get("capabilities"), dict):
        raise ValueError("capabilities必须是对象")
    for key in ("defaultInputModes", "defaultOutputModes"):
        if not strings(card.get(key)):
            raise ValueError(key + "必须是非空媒体类型列表")
    interfaces = card.get("supportedInterfaces")
    if not isinstance(interfaces, list) or not interfaces:
        raise ValueError("缺少supportedInterfaces")
    for item in interfaces:
        address = urlparse(item.get("url", ""))
        if address.scheme != "https" or not address.netloc:
            raise ValueError("接口地址必须是HTTPS绝对地址")
        if (item.get("protocolBinding"), item.get("protocolVersion")) != ("JSONRPC", "1.0"):
            raise ValueError("当前客户端需要JSONRPC 1.0接口")
    skills = card.get("skills")
    if not isinstance(skills, list) or not skills:
        raise ValueError("缺少skills")
    for skill in skills:
        if not all(text(skill.get(key)) for key in ("id", "name", "description")) or not strings(skill.get("tags")):
            raise ValueError("技能字段不完整")

def validate_parts(parts):
    if not isinstance(parts, list) or not parts:
        raise ValueError("parts不能为空")
    for part in parts:
        # v1.0的内容联合字段，不再使用旧kind判别字段。
        if not isinstance(part, dict) or len(set(part) & {"text", "raw", "url", "data"}) != 1:
            raise ValueError("每个Part只能有一种内容")
        if "text" in part and not isinstance(part["text"], str):
            raise ValueError("text必须是字符串")

def validate_message(message, task=None):
    if not text(message.get("messageId")) or message.get("role") not in {"ROLE_USER", "ROLE_AGENT"}:
        raise ValueError("消息编号或角色不合法")
    validate_parts(message.get("parts"))
    if message["role"] == "ROLE_AGENT" and not text(message.get("contextId")):
        raise ValueError("服务端消息缺少contextId")
    if task is not None:
        if message.get("taskId") not in (None, task["id"]):
            raise ValueError("taskId不匹配")
        if message.get("contextId") not in (None, task.get("contextId")):
            raise ValueError("contextId不匹配")

def transition(task, state):
    current = task["status"]["state"]
    if state not in STATES:
        raise ValueError("未知任务状态")
    if current in TERMINAL:
        raise ValueError("终态任务不能重新运行")
    # 允许边由这份退款应用定义；协议定义状态含义。
    if state not in NEXT.get(current, set()):
        raise ValueError("应用不允许此状态变化")
    result = deepcopy(task)
    result["status"]["state"] = state
    return result

def main():
    validate_card(CARD)
    validate_message(MESSAGE)
    interface = CARD["supportedInterfaces"][0]
    print("名片接口：" + interface["protocolBinding"] + " " + interface["protocolVersion"])
    print("用户消息：" + MESSAGE["parts"][0]["text"])
    task = {"id": "T-17", "contextId": "C-17", "status": {"state": "TASK_STATE_SUBMITTED"}}
    print("任务：" + task["id"] + " / " + task["contextId"])
    for state in ["TASK_STATE_WORKING", "TASK_STATE_INPUT_REQUIRED",
                  "TASK_STATE_WORKING", "TASK_STATE_COMPLETED"]:
        task = transition(task, state)
        print("状态：" + state)
    try:
        transition(task, "TASK_STATE_WORKING")
    except ValueError as error:
        print("再次运行：" + str(error))

if __name__ == "__main__":
    main()

ls
python -X utf8 a2a_demo.py

名片接口：JSONRPC 1.0
用户消息：查询O-17的退款规则
任务：T-17 / C-17
状态：TASK_STATE_WORKING
状态：TASK_STATE_INPUT_REQUIRED
状态：TASK_STATE_WORKING
状态：TASK_STATE_COMPLETED
再次运行：终态任务不能重新运行

python -X utf8 -m unittest -v test_a2a_demo.py