把它跑起来：一个 A2A server，一个 client，端到端

# Python 3.10+
python -m venv .venv && source .venv/bin/activate

# 锁定版本，避免装到 v0.3.x 的旧 API（见下方预防说明）
pip install "a2a-sdk==1.1.0"

# 自检：导入不报错、且版本是 1.1.0 = 装对了
python -c "import a2a; from importlib.metadata import version; print(version('a2a-sdk'))"

# 未在本机验证 —— a2a-sdk 1.1.0（spec v1.0）
from a2a.helpers import get_message_text, new_task_from_user_message, new_text_part
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types.a2a_pb2 import TaskState


class UpperCaseExecutor(AgentExecutor):
    """把客户端发来的文本转成大写，作为一个 Artifact 返回。"""

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        # 1. 取出 / 新建一个有状态 Task（对应 01 章 §1.3 的 Task）
        if context.current_task:
            task = context.current_task
        else:
            task = new_task_from_user_message(context.message)
            await event_queue.enqueue_event(task)   # 把新建的 Task 投到事件队列

        # 2. 用 TaskUpdater 推进状态：先置 WORKING
        updater = TaskUpdater(
            event_queue=event_queue,
            task_id=task.id,
            context_id=task.context_id,
        )
        await updater.update_status(state=TaskState.TASK_STATE_WORKING)

        # 3. 从入站 Message 抽出纯文本，干活（这里就是 .upper()）
        query = get_message_text(context.message)
        result = query.upper() if query else "(空输入)"

        # 4. 把结果包成 Artifact（对应 01 章 §1.3 的 Artifact = 任务产出）
        await updater.add_artifact(
            parts=[new_text_part(text=result, media_type="text/plain")],
            name="uppercased",
        )

        # 5. 终态：COMPLETED（对应 02 章 §2.5 生命周期的终态）
        await updater.update_status(state=TaskState.TASK_STATE_COMPLETED)

    async def cancel(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        # 本例不支持取消；真实长任务应在此置 TASK_STATE_CANCELED
        raise NotImplementedError("Cancel is not supported.")

# 未在本机验证 —— a2a-sdk 1.1.0（spec v1.0）
import uvicorn
from starlette.applications import Starlette

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
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill

from upper_executor import UpperCaseExecutor

PORT = 9999
BASE_URL = f"http://127.0.0.1:{PORT}"

# ── 一项 skill：向别的 agent 声明本 agent 能干这件事（对应 01 章 §1.2 的 skills）──
skill = AgentSkill(
    id="uppercase",
    name="Uppercase Text",
    description="把输入的纯文本整体转成大写。",
    input_modes=["text/plain"],
    output_modes=["text/plain"],
    tags=["text", "demo"],
    examples=["hello a2a", "make this loud"],
)

# ── Agent Card：对外的一纸能力声明（对应 01 章 §1.2、02 章 §2.2）──
agent_card = AgentCard(
    name="Uppercase Agent",
    description="一个把文本转大写的最小 A2A agent。",
    version="1.0.0",
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    # capabilities：声明支持哪些可选能力。这里先不开流式（3.3 让你决定要不要开）
    capabilities=AgentCapabilities(streaming=False),
    # supported_interfaces：声明走哪种传输、服务端点在哪（这里 JSON-RPC）
    supported_interfaces=[
        AgentInterface(protocol_binding="JSONRPC", url=BASE_URL),
    ],
    skills=[skill],
)

# ── 把 executor、task store、card 接到请求处理器上 ──
handler = DefaultRequestHandler(
    agent_executor=UpperCaseExecutor(),
    task_store=InMemoryTaskStore(),
    agent_card=agent_card,
)

# ── 生成路由：一组发布 Agent Card，一组处理 JSON-RPC 的 SendMessage 等 ──
routes = []
routes.extend(create_agent_card_routes(agent_card))   # GET /.well-known/agent-card.json
routes.extend(create_jsonrpc_routes(handler, "/"))    # POST / —— JSON-RPC 入口

app = Starlette(routes=routes)

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=PORT)

# 未在本机验证 —— a2a-sdk 1.1.0（spec v1.0）
import asyncio
import httpx

from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.helpers import get_stream_response_text, new_text_message
from a2a.types.a2a_pb2 import Role, SendMessageRequest

BASE_URL = "http://127.0.0.1:9999"


async def main() -> None:
    async with httpx.AsyncClient() as httpx_client:
        # 1. 发现：去固定路径 GET 那张 Agent Card（对应 01 §1.2 / 02 §2.2）
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=BASE_URL)
        agent_card = await resolver.get_agent_card()   # GET /.well-known/agent-card.json
        print("发现 skill:", [s.id for s in agent_card.skills])

        # 2. 用 Card 造一个 client。streaming=False 与本例 server 的声明一致
        config = ClientConfig(streaming=False, httpx_client=httpx_client)
        client = ClientFactory(config).create(agent_card)

        # 3. 发：构造一个 user 角色的 Message，调 send_message
        #    注意：Python 方法名是 snake_case send_message，
        #    但它在 wire 上发出的 JSON-RPC "method" 是 PascalCase "SendMessage"
        message = new_text_message("hello a2a", role=Role.ROLE_USER)
        request = SendMessageRequest(message=message)

        # 4. 收：send_message 返回异步迭代器，逐块收回 Task / Artifact
        async for chunk in client.send_message(request):
            print("收到:", get_stream_response_text(chunk))

        await client.close()


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

# 终端 1：起 server（监听 127.0.0.1:9999）
python server.py

# 终端 2：先手动看一眼那张 Card（发现这一步本质就是个 HTTP GET）
curl -s http://127.0.0.1:9999/.well-known/agent-card.json | head -c 400
echo

# 终端 2：再跑 client
python client.py

# 未在本机验证 —— 在 3.2 基础上抠掉 3 个 schema 决策
from a2a.helpers import get_message_text, new_task_from_user_message, new_text_part
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types.a2a_pb2 import TaskState

# ─────────────────────────────────────────────
# 决策 (a) · Card 里要不要声明 capabilities.streaming = True？
#   3.2 声明的是 False。如果这个 agent 要逐字吐结果（像 LLM 流式输出），
#   该把它改成 True 吗？改了之后，server 端 execute 里要相应多做什么？
#   （这个开关在 server.py 的 AgentCapabilities(...) 里，见下方答案）
# ─────────────────────────────────────────────

class UpperCaseExecutor(AgentExecutor):
    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        if context.current_task:
            task = context.current_task
        else:
            task = new_task_from_user_message(context.message)
            await event_queue.enqueue_event(task)

        updater = TaskUpdater(
            event_queue=event_queue,
            task_id=task.id,
            context_id=task.context_id,
        )
        await updater.update_status(state=TaskState.TASK_STATE_WORKING)

        query = get_message_text(context.message)

        # ─────────────────────────────────────────────
        # 决策 (c) · input-required 中途态怎么处理？
        #   如果客户端发来空文本（没东西可转大写），该怎么办？
        #   直接 FAILED？还是把任务暂停成 INPUT_REQUIRED 等它补一句？
        # ─────────────────────────────────────────────
        if not query:
            # await updater.update_status(state=???)   # 该置成哪个状态？然后 return 让出控制权？
            return

        result = query.upper()

        # ─────────────────────────────────────────────
        # 决策 (b) · 结果怎么包成 Artifact？
        #   下面这行只发了一个 text Part。如果还想附一份「原文 + 大写」的
        #   结构化对照（JSON），该往 parts 里加什么 kind 的 Part？
        # ─────────────────────────────────────────────
        await updater.add_artifact(
            parts=[new_text_part(text=result, media_type="text/plain")],  # 还要再加一个 Part 吗？
            name="uppercased",
        )

        await updater.update_status(state=TaskState.TASK_STATE_COMPLETED)

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise NotImplementedError("Cancel is not supported.")