《AI Agent 高级编程》8.1 A2A 协议：Agent Card/Task/消息，起两个本地 Agent 通信

"""A2A 最小服务端（Python 标准库）：暴露 Agent Card 与 message/send 任务端点。"""
import json
import uuid
import hashlib
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

PORT = 8741

AGENT_CARD = {
    "protocolVersion": "0.3.0",
    "name": "ResearchCopilot-Py",
    "description": "本地研究助理 Agent，提供一个确定性的文本统计技能。",
    "url": f"http://127.0.0.1:{PORT}/",
    "version": "0.1.0",
    "capabilities": {"streaming": False, "pushNotifications": False},
    "defaultInputModes": ["text/plain"],
    "defaultOutputModes": ["application/json"],
    "skills": [
        {
            "id": "text-stats",
            "name": "文本统计",
            "description": "对输入文本做确定性统计（字符数/词数/指纹），返回 JSON Artifact。",
            "tags": ["text", "stats"],
            "examples": ["统计这段话的字符数与词数"],
        }
    ],
}


def run_skill(text: str) -> dict:
    """确定性的技能实现：不含任何随机与外部调用。"""
    words = text.split()
    fingerprint = hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]
    return {
        "chars": len(text),
        "words": len(words),
        "fingerprint": fingerprint,
        "summary": f"共 {len(text)} 字符、{len(words)} 个词，指纹 {fingerprint}",
    }


def handle_message_send(params: dict) -> dict:
    """把 A2A 的 Message 转成内部技能调用，再包成 Task 返回。"""
    message = params["message"]
    parts = message.get("parts", [])
    text = "".join(p.get("text", "") for p in parts if p.get("kind") == "text")
    task_id = message.get("taskId") or uuid.uuid4().hex
    result = run_skill(text)
    return {
        "id": task_id,
        "contextId": message.get("contextId", uuid.uuid4().hex),
        "status": {"state": "completed", "timestamp": "2026-10-07T18:00:00+08:00"},
        "history": [message],
        "artifacts": [
            {
                "artifactId": uuid.uuid4().hex[:8],
                "name": "text-stats.json",
                "parts": [{"kind": "data", "data": result}],
            }
        ],
    }


class Handler(BaseHTTPRequestHandler):
    def _json(self, code: int, payload: dict):
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == "/.well-known/agent-card.json":
            self._json(200, AGENT_CARD)
        else:
            self._json(404, {"error": "not found"})

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        req = json.loads(self.rfile.read(length) or b"{}")
        if req.get("method") != "message/send":
            self._json(200, {"jsonrpc": "2.0", "id": req.get("id"),
                             "error": {"code": -32601, "message": "method not found"}})
            return
        task = handle_message_send(req["params"])
        self._json(200, {"jsonrpc": "2.0", "id": req.get("id"), "result": task})

    def log_message(self, *args):
        pass


if __name__ == "__main__":
    server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
    print(f"Agent Card: http://127.0.0.1:{PORT}/.well-known/agent-card.json")
    server.serve_forever()

"""A2A 客户端：先取 Agent Card 发现技能，再发起一次 message/send 任务。"""
import json
import urllib.request

BASE = "http://127.0.0.1:8741"


def get_json(url: str) -> dict:
    with urllib.request.urlopen(url, timeout=5) as resp:
        return json.loads(resp.read().decode("utf-8"))


def post_json(url: str, payload: dict) -> dict:
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(url, data=data,
                                headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=5) as resp:
        return json.loads(resp.read().decode("utf-8"))


card = get_json(f"{BASE}/.well-known/agent-card.json")
print("发现 Agent:", card["name"], "| 协议", card["protocolVersion"])
print("技能:", [s["id"] for s in card["skills"]])

rpc = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "message/send",
    "params": {
        "message": {
            "role": "user",
            "messageId": "m-1",
            "parts": [{"kind": "text", "text": "A2A 让 Agent 之间直接对话"}],
        }
    },
}
task = post_json(card["url"], rpc)["result"]
print("任务状态:", task["status"]["state"])
print("Artifact:", json.dumps(task["artifacts"][0]["parts"][0]["data"],
                             ensure_ascii=False))

{"protocolVersion": "0.3.0", "name": "ResearchCopilot-Py", "description": "本地研究助理 Agent，提供一个确定性的文本统计技能。", "url": "http://127.0.0.1:8741/", "version": "0.1.0", "capabilities": {"streaming": false, "pushNotifications": false}, "defaultInputModes": ["text/plain"], "defaultOutputModes": ["application/json"], "skills": [{"id": "text-stats", "name": "文本统计", "description": "对输入文本做确定性统计（字符数/词数/指纹），返回 JSON Artifact。", "tags": ["text", "stats"], "examples": ["统计这段话的字符数与词数"]}]}

发现 Agent: ResearchCopilot-Py | 协议 0.3.0
技能: ['text-stats']
任务状态: completed
Artifact: {"chars": 18, "words": 4, "fingerprint": "30ec2678a79e", "summary": "共 18 字符、4 个词，指纹 30ec2678a79e"}