在 0.3 环境查询与维护已有任务

为什么旧对象必须由旧协议解码

线上遗留服务仍用 tasks/get，维护者首先要确认订单是否完成。升级 SDK 后的枚举和响应包装变化，不应改变这条业务判断。本节固定 Python 3.12、a2a-sdk 0.3.0，独立目录建立环境，不覆盖第一章的 1.1.0 环境：

python3.12 -m venv .venv-old
source .venv-old/bin/activate
python -m pip install 'a2a-sdk==0.3.0' 'httpx==0.28.1' 'starlette==1.6.0' 'sse-starlette==3.4.11'
python -m pip freeze > requirements-old.lock

保存 old_task.py，运行 python old_task.py。实际旧 SDK 的服务应用与客户端通过 HTTP 序列化相连；ASGITransport 省去 TCP 监听，不替换请求处理器。任务仓库预置已有订单，模拟维护系统已有工作；没有启动模型或声称 worker 自动推进状态。两次 save 表示外部工作者分别写入 working 和 completed。

import asyncio,httpx
from a2a.client import A2AClient
from a2a.server.apps import A2AStarletteApplication
from a2a.server.agent_execution import AgentExecutor
from a2a.server.tasks import InMemoryTaskStore
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.types import AgentCard,AgentCapabilities,Task,TaskStatus,TaskState,GetTaskRequest,TaskQueryParams,GetTaskSuccessResponse,JSONRPCErrorResponse
class ReadOnly(AgentExecutor):
    async def execute(self,context,event_queue):raise NotImplementedError('maintenance read only')
    async def cancel(self,context,event_queue):raise NotImplementedError('maintenance read only')
async def main():
    store=InMemoryTaskStore()
    card=AgentCard(name='Archive',description='legacy task lookup',version='demo',url='http://old/',capabilities=AgentCapabilities(),default_input_modes=['text'],default_output_modes=['text'],skills=[])
    handler=DefaultRequestHandler(agent_executor=ReadOnly(),task_store=store)
    app=A2AStarletteApplication(agent_card=card,http_handler=handler).build()
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app)) as http:
        client=A2AClient(httpx_client=http,agent_card=card)
        for state in (TaskState.working,TaskState.completed):
            await store.save(Task(id='order-A',context_id='c-A',status=TaskStatus(state=state)))
            response=await client.get_task(GetTaskRequest(id='query',params=TaskQueryParams(id='order-A')))
            assert isinstance(response.root,GetTaskSuccessResponse)
            task=response.root.result
            assert task.id=='order-A' and task.status.state==state
            archive=task.model_dump_json(by_alias=True)
            assert Task.model_validate_json(archive).status.state==state
            print(task.id,task.status.state.value)
        missing=await client.get_task(GetTaskRequest(id='absent',params=TaskQueryParams(id='missing')))
        assert isinstance(missing.root,JSONRPCErrorResponse)
        assert missing.root.error.code==-32001
        print('missing task rejected')
asyncio.run(main())

输出先是 order-A working，再是 order-A completed，最后是 missing task rejected。response.root 是成功或 JSON-RPC 错误的联合；必须先判成功，才能读取 result。错误不能变成一个空 Task，否则调用方可能把“没有任务”误判成“尚未完成”。context_id 对应 JSON 的 contextId，归档必须使用 by_alias=True；model_validate_json 用同一版本恢复。

A2AClient 在此旧版本已有弃用提示，它仍是存量可执行入口。维护旧服务时保留这份环境与行为断言；新主线用第一章 create_client 与 protobuf Task，状态是 TASK_STATE_WORKING/TASK_STATE_COMPLETED。迁移时保留业务订单 ID 和上下文，再映射状态；0.3 的 JSON-RPC 包装不可以当成 1.1 protobuf 对象。第三章前一篇讨论授权边界，本实验只展示受信本地存档，不能作为公网无鉴权任务查询服务部署。

练习

新增 failed 状态与 task ID order-B，要求相同旧客户端能读出 failed，且不会影响 order-A 的 completed。未知任务仍返回 -32001。请在保存与查询后断言这三项。

本章答案