身份、可靠性、路由与版本维护：练习答案

练习解答

from urllib.parse import urlsplit
def accept(url):
    p=urlsplit(url)
    return p.scheme=='https' and p.hostname in {'agents.example.com','backup.example.com'} and p.port in (None,443) and not p.username
assert accept('https://backup.example.com/rpc')
assert not accept('https://backup.example.com.evil.test')
assert not accept('https://agents.example.com:8080')
print('alternate route passed')

首跳合法不代表 Location 目标合法；每次重定向都需重新验证并应用网络限制。

旧版任务维护

在旧环境保存以下完整答案为 answer_old_task.py。两个任务分开保存和查询，断言读取身份与状态；未知 ID 仍走错误联合。

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)
        await store.save(Task(id='order-B',context_id='c-B',status=TaskStatus(state=TaskState.failed)))
        for task_id,state in [('order-A',TaskState.completed),('order-B',TaskState.failed)]:
            r=await client.get_task(GetTaskRequest(id=task_id,params=TaskQueryParams(id=task_id)))
            assert isinstance(r.root,GetTaskSuccessResponse) and r.root.result.status.state==state
        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())

输出保留 working、completed 和 missing task rejected；新增断言证明 order-B 的失败不会覆盖 order-A。