A2A 协议概览：从能力发现到任务交付

demo/
├── server.py
├── client.py
├── run_demo.sh
├── .venv/       # 本地创建的 Python 虚拟环境
├── server.log   # 一键脚本重定向的 Uvicorn 日志
└── server.pid   # 一键脚本运行期间记录的服务进程号

python3.13 -m venv .venv
.venv/bin/python -m pip install 'a2a-sdk==1.3.0' 'httpx==0.28.1' 'starlette==1.10.0' 'uvicorn==0.54.0'
.venv/bin/python -c 'from importlib.metadata import version; print("a2a-sdk", version("a2a-sdk"))'

from __future__ import annotations

from uuid import uuid4

from starlette.applications import Starlette

from a2a.server.agent_execution.agent_executor import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.agent_execution.simple_request_context_builder import SimpleRequestContextBuilder
from a2a.server.events.event_queue import EventQueue
from a2a.server.request_handlers.default_request_handler_v2 import DefaultRequestHandlerV2
from a2a.server.routes.agent_card_routes import create_agent_card_routes
from a2a.server.routes.jsonrpc_routes import create_jsonrpc_routes
from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
    Artifact,
    Message,
    Part,
    Task,
    TaskArtifactUpdateEvent,
    TaskState,
    TaskStatus,
    TaskStatusUpdateEvent,
)


class TripPlannerExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        task_id = context.task_id
        context_id = context.context_id
        if not task_id or not context_id:
            raise ValueError("The request handler did not assign task/context IDs")

        user_text = context.get_user_input().strip()
        if not context.current_task or context.current_task.status.state != TaskState.TASK_STATE_INPUT_REQUIRED:
            if context.related_tasks:
                await event_queue.enqueue_event(
                    Task(
                        id=task_id,
                        context_id=context_id,
                        status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
                    )
                )
                await self._complete_task(task_id, context_id, user_text, event_queue)
                return
            await event_queue.enqueue_event(
                Task(
                    id=task_id,
                    context_id=context_id,
                    status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
                )
            )
            await event_queue.enqueue_event(
                TaskStatusUpdateEvent(
                    task_id=task_id,
                    context_id=context_id,
                    status=TaskStatus(
                        state=TaskState.TASK_STATE_INPUT_REQUIRED,
                        message=Message(
                            parts=[Part(text="我可以规划北京往返上海的差旅。请补充预算上限，例如 8000 元。")],
                            role="ROLE_AGENT",
                        ),
                    ),
                )
            )
            return

        await self._complete_task(task_id, context_id, user_text, event_queue)

    async def _complete_task(
        self, task_id: str, context_id: str, budget: str, event_queue: EventQueue
    ) -> None:
        summary = f"行程草案：北京往返上海，预算上限 {budget}；建议先比较高铁与机票，再按住宿预算筛选酒店。"
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(
                task_id=task_id,
                context_id=context_id,
                status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
            )
        )
        await event_queue.enqueue_event(
            TaskArtifactUpdateEvent(
                task_id=task_id,
                context_id=context_id,
                artifact=Artifact(
                    artifact_id=str(uuid4()),
                    name="trip-plan",
                    description="根据补充预算生成的确定性演示结果",
                    parts=[Part(text=summary)],
                ),
                append=False,
                last_chunk=True,
            )
        )
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(
                task_id=task_id,
                context_id=context_id,
                status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
            )
        )

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        if context.task_id and context.context_id:
            await event_queue.enqueue_event(
                TaskStatusUpdateEvent(
                    task_id=context.task_id,
                    context_id=context.context_id,
                    status=TaskStatus(state=TaskState.TASK_STATE_CANCELED),
                )
            )


agent_card = AgentCard(
    name="Trip Planner Demo Agent",
    description="A deterministic A2A demo that asks for a budget and returns a trip-plan artifact.",
    supported_interfaces=[
        {
            "url": "http://127.0.0.1:8000/a2a",
            "protocol_binding": "JSONRPC",
            "protocol_version": "1.0",
        }
    ],
    version="1.0.0",
    capabilities=AgentCapabilities(streaming=False, push_notifications=False),
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    skills=[
        AgentSkill(
            id="trip-planning",
            name="Trip planning",
            description="Collects a budget and returns a simple trip-plan artifact.",
            tags=["travel", "planning"],
            examples=["Plan a Beijing to Shanghai business trip"],
            input_modes=["text/plain"],
            output_modes=["text/plain"],
        )
    ],
)

task_store = InMemoryTaskStore()
handler = DefaultRequestHandlerV2(
    agent_executor=TripPlannerExecutor(),
    task_store=task_store,
    agent_card=agent_card,
    request_context_builder=SimpleRequestContextBuilder(
        should_populate_referred_tasks=True,
        task_store=task_store,
    ),
)

app = Starlette(
    routes=[
        *create_agent_card_routes(agent_card),
        *create_jsonrpc_routes(handler, rpc_url="/a2a"),
    ]
)

from __future__ import annotations

import asyncio
import json

import httpx
from a2a.client import ClientConfig
from a2a.client.card_resolver import A2ACardResolver
from a2a.client.client_factory import create_client
from a2a.types import (
    GetTaskRequest,
    Message,
    Part,
    Role,
    SendMessageConfiguration,
    SendMessageRequest,
    Task,
    TaskState,
)


async def main() -> None:
    base_url = "http://127.0.0.1:8000"
    async with httpx.AsyncClient(trust_env=False) as http:
        card = await A2ACardResolver(http, base_url).get_agent_card()
    print("DISCOVERED_AGENT_CARD")
    print(json.dumps({"name": card.name, "version": card.version, "skills": [s.id for s in card.skills]}, ensure_ascii=False))

    client = await create_client(
        card,
        client_config=ClientConfig(
            streaming=False,
            polling=False,
            httpx_client=httpx.AsyncClient(trust_env=False),
        ),
    )
    try:
        first_message = Message(
            message_id="demo-initial-message",
            role=Role.ROLE_USER,
            parts=[Part(text="请规划一次北京往返上海的差旅。")],
        )
        first_request = SendMessageRequest(
            message=first_message,
            configuration=SendMessageConfiguration(return_immediately=False),
        )
        first_responses = [response async for response in client.send_message(first_request)]
        first_response = first_responses[0]
        first_task = first_response.task if first_response.HasField("task") else None
        if first_task is None:
            raise RuntimeError(f"Expected a Task, got: {first_responses!r}")
        print("FIRST_RESPONSE")
        print(json.dumps({"taskId": first_task.id, "contextId": first_task.context_id, "state": TaskState.Name(first_task.status.state), "statusMessage": [p.text for p in first_task.status.message.parts]}, ensure_ascii=False))

        follow_up = Message(
            message_id="demo-budget-follow-up",
            context_id=first_task.context_id,
            role=Role.ROLE_USER,
            parts=[Part(text="预算上限 8000 元。")],
            reference_task_ids=[first_task.id],
        )
        follow_up_request = SendMessageRequest(
            message=follow_up,
            configuration=SendMessageConfiguration(return_immediately=False),
        )
        final_responses = [response async for response in client.send_message(follow_up_request)]
        final_response = final_responses[0]
        final_task = final_response.task if final_response.HasField("task") else None
        if final_task is None:
            raise RuntimeError(f"Expected a Task, got: {final_responses!r}")
        artifacts = [
            {"name": artifact.name, "text": "\n".join(part.text for part in artifact.parts if part.text)}
            for artifact in final_task.artifacts
        ]
        print("FINAL_RESPONSE")
        print(json.dumps({"taskId": final_task.id, "contextId": final_task.context_id, "state": TaskState.Name(final_task.status.state), "artifacts": artifacts}, ensure_ascii=False))

        first_persisted = await client.get_task(GetTaskRequest(id=first_task.id))
        if TaskState.Name(first_persisted.status.state) != "TASK_STATE_INPUT_REQUIRED":
            raise AssertionError(f"First task state changed unexpectedly: {TaskState.Name(first_persisted.status.state)}")
        persisted = await client.get_task(GetTaskRequest(id=final_task.id))
        if TaskState.Name(persisted.status.state) != "TASK_STATE_COMPLETED":
            raise AssertionError(f"Follow-up task did not persist in completed state: {TaskState.Name(persisted.status.state)}")
        print("GET_TASK")
        print(json.dumps({"firstTaskState": TaskState.Name(first_persisted.status.state), "followUpTaskId": persisted.id, "followUpState": TaskState.Name(persisted.status.state), "artifactCount": len(persisted.artifacts)}, ensure_ascii=False))
    finally:
        await client.close()


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

.venv/bin/python -m uvicorn server:app --host 127.0.0.1 --port 8000

.venv/bin/python client.py

sh run_demo.sh

#!/bin/sh
set -eu

SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
PYTHON="$SCRIPT_DIR/.venv/bin/python"
LOG="$SCRIPT_DIR/server.log"
PID_FILE="$SCRIPT_DIR/server.pid"

if [ ! -x "$PYTHON" ]; then
  echo "Missing virtualenv Python: $PYTHON" >&2
  exit 1
fi

"$PYTHON" -c 'import a2a, httpx, starlette, uvicorn' || {
  echo "Missing runtime dependency. Install the documented requirements first." >&2
  exit 1
}

"$PYTHON" -m uvicorn server:app --app-dir "$SCRIPT_DIR" --host 127.0.0.1 --port 8000 >"$LOG" 2>&1 &
SERVER_PID=$!
printf '%s\n' "$SERVER_PID" > "$PID_FILE"
cleanup() {
  kill "$SERVER_PID" 2>/dev/null || true
  wait "$SERVER_PID" 2>/dev/null || true
  rm -f "$PID_FILE"
}
trap cleanup EXIT INT TERM

READY=0
for _ in $(seq 1 50); do
  if curl --silent --fail http://127.0.0.1:8000/.well-known/agent-card.json >/dev/null; then
    READY=1
    break
  fi
  sleep 0.1
done
if [ "$READY" -ne 1 ]; then
  echo "Server did not become ready. See $LOG" >&2
  exit 1
fi

"$PYTHON" "$SCRIPT_DIR/client.py"

curl --noproxy '*' http://127.0.0.1:8000/.well-known/agent-card.json

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "method": "SendMessage",
  "params": {
    "message": {
      "messageId": "demo-budget-follow-up",
      "contextId": "577ba9c0-7ec0-4674-8f09-26b37c63840f",
      "role": "ROLE_USER",
      "parts": [{"text": "预算上限 8000 元。"}],
      "referenceTaskIds": ["a1e6b39c-2481-45c5-914d-120a55e5c617"]
    },
    "configuration": {"returnImmediately": false}
  }
}

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "method": "GetTask",
  "params": {"id": "1f93834d-99a6-4302-b1e2-2cbcdc1d8803"}
}

.venv/bin/python client.py
DISCOVERED_AGENT_CARD
{"name": "Trip Planner Demo Agent", "version": "1.0.0", "skills": ["trip-planning"]}
FIRST_RESPONSE
{"taskId": "a3c0aa68-30f9-48f1-b09f-6237e44c533d", "contextId": "70ac763f-ed9d-4d77-921c-ab02f14d495c", "state": "TASK_STATE_INPUT_REQUIRED", "statusMessage": ["我可以规划北京往返上海的差旅。请补充预算上限，例如 8000 元。"]}
FINAL_RESPONSE
{"taskId": "dcc0ee97-a763-4ec2-a788-2039cd2b0053", "contextId": "70ac763f-ed9d-4d77-921c-ab02f14d495c", "state": "TASK_STATE_COMPLETED", "artifacts": [{"name": "trip-plan", "text": "行程草案：北京往返上海，预算上限 预算上限 8000 元。；建议比较高铁与机票，再按住宿预算筛选酒店。"}]}
GET_TASK
{"firstTaskState": "TASK_STATE_INPUT_REQUIRED", "followUpTaskId": "dcc0ee97-a763-4ec2-a788-2039cd2b0053", "followUpState": "TASK_STATE_COMPLETED", "artifactCount": 1}

follow_up = Message(
    message_id="demo-budget-follow-up",
    task_id=first_task.id,
    context_id=first_task.context_id,
    role=Role.ROLE_USER,
    parts=[Part(text="预算上限 8000 元。")],
)

persisted = await client.get_task(GetTaskRequest(id=first_task.id))
assert TaskState.Name(persisted.status.state) == "TASK_STATE_COMPLETED"
assert len(persisted.artifacts) == 1