搭建一个A2A Server

我们将用约 100 行 Python 写出第一个完整的 A2A 系统：一个不需要 LLM 的掷骰子 Agent（Server）+ 一个会做能力发现和流式调用的 Client。

所有代码和输出均已在 Python 3.11 + a2a-sdk 1.3.0 下实际跑通。

环境准备

mkdir a2a-demo && cd a2a-demo
python3 -m venv venv
./venv/bin/pip install "a2a-sdk[http-server]" uvicorn httpx
# 本教程验证版本：a2a-sdk 1.3.0

架构

┌──────────────────┐    ① GET /.well-known/agent-card.json    ┌──────────────────┐
│   dice_client.py │ ─────────────────────────────────────────>│  dice_server.py  │
│   (A2A Client)   │ <───────────── AgentCard ────────────────│  (A2A Server)    │
│                  │                                           │                  │
│                  │    ② SendStreamingMessage("roll 3d6")     │  DiceAgent       │
│                  │ ─────────────────────────────────────────>│  Executor        │
│                  │ <══ Task / 状态更新 / 产物 (SSE 流式) ════│  (纯 Python      │
└──────────────────┘                                           │   逻辑,无 LLM)   │
                                                               └──────────────────┘

Server 端只需要实现一个 AgentExecutor（业务逻辑），SDK 负责其余一切： HTTP 路由、JSON-RPC 解析、任务状态管理、SSE 流式。

Server 代码

创建一个dice_server.py：

"""实战一：最小 A2A Server —— 掷骰子 Agent（不需要 LLM）。

运行: ./venv/bin/python dice_server.py   (监听 http://localhost:9999)
"""

import random
import re

import uvicorn
from starlette.applications import Starlette

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentInterface,
    AgentSkill,
    Part,
    Task,
    TaskState,
    TaskStatus,
)


class DiceAgentExecutor(AgentExecutor):
    """核心业务逻辑：解析 "NdM" 格式并掷骰子。"""

    async def execute(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        # 第一个事件必须是 Task 对象本身（状态 submitted）
        if context.current_task is None:
            task = Task(
                id=context.task_id,
                context_id=context.context_id,
                status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
                history=[context.message] if context.message else [],
            )
            await event_queue.enqueue_event(task)

        # TaskUpdater 是向事件队列发布任务状态/产物的辅助类
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.start_work()    # 状态 -> working

        user_text = context.get_user_input()  # 取出用户消息里的文本
        m = re.search(r"(\d+)\s*[dD]\s*(\d+)", user_text)
        n, faces = (int(m.group(1)), int(m.group(2))) if m else (1, 6)
        n = min(n, 100)
        rolls = [random.randint(1, faces) for _ in range(n)]
        result = f"掷 {n} 个 {faces} 面骰子: {rolls}，总和 = {sum(rolls)}"

        # 以 Artifact（任务产物）形式返回结果
        await updater.add_artifact(
            [Part(text=result)], name="dice-result"
        )
        await updater.complete()      # 状态 -> completed（终态）

    async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.cancel()


def build_agent_card() -> AgentCard:
    """Agent Card：对外公开的“名片”，发布在 /.well-known/agent-card.json"""
    return AgentCard(
        name="Dice Agent",
        description="一个会掷骰子的极简 A2A Agent，输入如 '2d6' 即掷 2 个 6 面骰。",
        version="1.0.0",
        supported_interfaces=[
            AgentInterface(
                url="http://localhost:9999/",
                protocol_binding="JSONRPC",
                protocol_version="1.0",
            )
        ],
        capabilities=AgentCapabilities(streaming=True),
        default_input_modes=["text/plain"],
        default_output_modes=["text/plain"],
        skills=[
            AgentSkill(
                id="roll_dice",
                name="掷骰子",
                description="按 NdM 格式掷 N 个 M 面骰子并返回结果",
                tags=["dice", "game"],
                examples=["roll 2d6", "掷 3d20"],
            )
        ],
    )


def main() -> None:
    card = build_agent_card()
    handler = DefaultRequestHandler(
        agent_executor=DiceAgentExecutor(),
        task_store=InMemoryTaskStore(),
        agent_card=card,
    )
    app = Starlette(
        routes=[
            *create_agent_card_routes(card),          # GET /.well-known/agent-card.json
            *create_jsonrpc_routes(handler, rpc_url="/"),  # POST /  (JSON-RPC 2.0)
        ]
    )
    uvicorn.run(app, host="0.0.0.0", port=9999, log_level="warning")


if __name__ == "__main__":
    main()

关键点：

  • AgentCard：声明身份、端点（supported_interfaces）、技能、流式能力；
  • AgentExecutor.execute()：业务逻辑入口。事件发布顺序是铁律—— 先 Task，再状态更新，再产物，最后终态；
  • TaskUpdater：封装了状态事件的辅助类（start_work/add_artifact/ complete/failed/requires_input…）；
  • DefaultRequestHandler + create_jsonrpc_routes：把 Executor 接到 JSON-RPC 端点上；create_agent_card_routes 负责发布名片。

运行这个脚本：

./venv/bin/python dice_server.py   # 终端 1

然后访问浏览器的 http://localhost:9999/.well-known/agent-card.json 路径：

image-20261112220547057

Client 代码

创建一个dice_client.py：

"""实战一：A2A Client —— 发现 Agent、发送消息、接收流式结果。

运行: ./venv/bin/python dice_client.py
"""

import asyncio
import uuid

from a2a.client import A2ACardResolver, create_client
import httpx
from a2a.types import Message, Part, Role, SendMessageRequest, TaskState

BASE_URL = "http://localhost:9999"


async def main() -> None:
    # 1) 能力发现：拉取 /.well-known/agent-card.json
    async with httpx.AsyncClient() as http:
        resolver = A2ACardResolver(http, BASE_URL)
        card = await resolver.get_agent_card()
    print(f"[发现] Agent: {card.name} v{card.version}")
    print(f"[发现] 描述: {card.description}")
    print(f"[发现] 技能: {[s.name for s in card.skills]}")
    print(f"[发现] 支持流式: {card.capabilities.streaming}")

    # 2) 创建客户端并发送消息（SDK 自动选择 JSON-RPC binding）
    client = await create_client(card)
    request = SendMessageRequest(
        message=Message(
            message_id=str(uuid.uuid4()),
            role=Role.ROLE_USER,
            parts=[Part(text="帮我 roll 3d6")],
        )
    )

    # 3) 迭代流式响应（Task / status_update / artifact_update 事件）
    print("\n[发送] 帮我 roll 3d6")
    async for event in client.send_message(request):
        if event.HasField("task"):
            print(f"[事件] Task 创建: id={event.task.id[:8]}... "
                  f"状态={TaskState.Name(event.task.status.state)}")
        elif event.HasField("status_update"):
            su = event.status_update
            print(f"[事件] 状态更新: {TaskState.Name(su.status.state)}")
        elif event.HasField("artifact_update"):
            au = event.artifact_update
            text = "".join(p.text for p in au.artifact.parts if p.HasField("text"))
            print(f"[事件] 产物 '{au.artifact.name}': {text}")

    await client.close()


if __name__ == "__main__":
    asyncio.run(main())

关键点：

  • A2ACardResolver：拉取并解析 Agent Card（能力发现）；
  • create_client(card)：根据卡片自动选择传输绑定（这里是 JSON-RPC）；
  • client.send_message() 返回异步迭代器：Server 支持流式时走 SSE， 不支持时 SDK 自动降级为单次响应——Client 代码不用改。

运行 Client：

./venv/bin/python dice_client.py

Client 输出：

image-20261112220816933

也可以不用 SDK，直接 curl 裸调 JSON-RPC：

curl -s http://localhost:9999/ -X POST \
  -H 'Content-Type: application/json' -H 'A2A-Version: 1.0' \
  -d '{"jsonrpc":"2.0","id":1,"method":"SendMessage",
       "params":{"message":{"messageId":"msg-001","role":"ROLE_USER",
                 "parts":[{"text":"roll 2d6"}]}}}'

返回：

image-20261112220925746