补参、持久查询与取消顺序
等待输入也是可以保存的任务
第一节Echo太快，无法观察任务停在中间。下面仍使用a2a-sdk 1.1.0，增加SQLite保存真正Task protobuf，并让缺SKU的任务停在INPUT_REQUIRED。应用只实现库存查询，不调用模型。保存durable.py运行；ASGITransport通过真实SDK的JSON-RPC HTTP序列化访问Starlette，但在本进程内传输，无TCP端口。这与第一章真实网络握手分别验证。
python
import asyncio,sqlite3,subprocess,sys
from tempfile import TemporaryDirectory
from pathlib import Path
import httpx
from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from a2a.server.tasks import TaskStore,TaskUpdater
from a2a.server.agent_execution import AgentExecutor
from a2a.server.context import ServerCallContext
from a2a.server.routes.common import ServerCallContextBuilder
from a2a.server.routes import create_jsonrpc_routes,create_agent_card_routes
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.client import create_client,ClientConfig
from a2a.helpers import new_text_message,new_text_part,new_task_from_user_message
from a2a.types import *
from a2a.utils.errors import TaskNotFoundError,TaskNotCancelableError
class Store(TaskStore):
def __init__(self,path):
self.db=sqlite3.connect(path)
self.db.execute('CREATE TABLE IF NOT EXISTS tasks(id TEXT PRIMARY KEY,owner TEXT,body BLOB)')
async def save(self,task,context):
owner=self.db.execute('SELECT owner FROM tasks WHERE id=?',(task.id,)).fetchone()
if owner and owner[0]!=context.state['owner']:raise PermissionError('owner mismatch')
with self.db:self.db.execute('INSERT INTO tasks VALUES(?,?,?) ON CONFLICT(id) DO UPDATE SET body=excluded.body',
(task.id,context.state['owner'],task.SerializeToString()))
async def get(self,task_id,context):
row=self.db.execute('SELECT body FROM tasks WHERE id=? AND owner=?',(task_id,context.state['owner'])).fetchone()
return Task.FromString(row[0]) if row else None
async def delete(self,task_id,context):
with self.db:self.db.execute('DELETE FROM tasks WHERE id=? AND owner=?',(task_id,context.state['owner']))
async def list(self,params,context):
rows=self.db.execute('SELECT body FROM tasks WHERE owner=?',(context.state['owner'],)).fetchall()
tasks=[Task.FromString(r[0]) for r in rows]
return ListTasksResponse(tasks=tasks,total_size=len(tasks),page_size=len(tasks))
class Identity(ServerCallContextBuilder):
def build(self,request):
tenant={'Bearer local-a':'a','Bearer local-b':'b'}.get(request.headers.get('authorization'))
if not tenant:raise HTTPException(401)
return ServerCallContext(state={"headers":dict(request.headers),"owner":tenant})
class Order(AgentExecutor):
async def execute(self,context,event_queue):
task=context.current_task or new_task_from_user_message(context.message)
if not context.current_task:await event_queue.enqueue_event(task)
u=TaskUpdater(event_queue=event_queue,task_id=task.id,context_id=task.context_id)
if context.get_user_input()!='sku:A':
await u.update_status(state=TaskState.TASK_STATE_INPUT_REQUIRED,
message=new_text_message('请补充sku:A',role=Role.ROLE_AGENT))
return
await u.update_status(state=TaskState.TASK_STATE_WORKING)
await u.add_artifact(artifact_id='stock',parts=[new_text_part(text='A:',media_type='text/plain')],append=False,last_chunk=False)
await u.add_artifact(artifact_id='stock',parts=[new_text_part(text='3',media_type='text/plain')],append=True,last_chunk=True)
await u.update_status(state=TaskState.TASK_STATE_COMPLETED)
async def cancel(self,context,event_queue):
u=TaskUpdater(event_queue=event_queue,task_id=context.task_id,context_id=context.context_id)
await u.update_status(state=TaskState.TASK_STATE_CANCELED)
def app_for(path):
store=Store(path)
card=AgentCard(name='Order',description='durable protocol lab',version='1',
default_input_modes=['text/plain'],default_output_modes=['text/plain'],
capabilities=AgentCapabilities(streaming=True),
supported_interfaces=[AgentInterface(protocol_binding='JSONRPC',url='http://lab/',protocol_version='1.0')],
skills=[AgentSkill(id='stock',name='stock',description='needs sku',tags=['stock'])])
handler=DefaultRequestHandler(agent_executor=Order(),task_store=store,agent_card=card)
return Starlette(routes=create_agent_card_routes(card)+create_jsonrpc_routes(handler,'/',context_builder=Identity())),card,store
async def client_for(app,card,owner):
http=httpx.AsyncClient(transport=httpx.ASGITransport(app=app),headers={'Authorization':'Bearer local-'+owner})
c=await create_client(agent=card,client_config=ClientConfig(streaming=False,httpx_client=http))
return c,http
async def send(client,text,task=None):
msg=new_text_message(text,role=Role.ROLE_USER)
if task:msg.task_id=task.id;msg.context_id=task.context_id
final=None
async for result in client.send_message(SendMessageRequest(message=msg)):
if result.HasField('task'):final=result.task
assert final is not None
return final
async def main():
with TemporaryDirectory() as d:
path=Path(d)/'tasks.db'
app,card,store=app_for(path);c,http=await client_for(app,card,'a')
waiting=await send(c,'stock please')
assert waiting.status.state==TaskState.TASK_STATE_INPUT_REQUIRED
await c.close();await http.aclose();store.db.close()
# A fresh interpreter proves this snapshot is not only in Python memory.
check="import sqlite3,sys;from a2a.types import Task; row=sqlite3.connect(sys.argv[1]).execute('SELECT body FROM tasks WHERE id=?',(sys.argv[2],)).fetchone();assert Task.FromString(row[0]).id==sys.argv[2]"
subprocess.run([sys.executable,'-c',check,str(path),waiting.id],check=True)
# Destroy service/store objects, then reconstruct using the same persisted DB.
app,card,store=app_for(path);c,http=await client_for(app,card,'a')
restored=await c.get_task(GetTaskRequest(id=waiting.id))
assert restored.context_id==waiting.context_id
done=await send(c,'sku:A',restored)
assert done.id==waiting.id and done.status.state==TaskState.TASK_STATE_COMPLETED
assert ''.join(p.text for p in done.artifacts[0].parts)=='A:3'
try:await c.cancel_task(CancelTaskRequest(id=done.id))
except TaskNotCancelableError:pass
assert (await c.get_task(GetTaskRequest(id=done.id))).status.state==TaskState.TASK_STATE_COMPLETED
pending=await send(c,'another task')
cancelled=await c.cancel_task(CancelTaskRequest(id=pending.id))
assert cancelled.status.state==TaskState.TASK_STATE_CANCELED
other,other_http=await client_for(app,card,'b')
try:await other.get_task(GetTaskRequest(id=done.id))
except TaskNotFoundError:pass
else:raise AssertionError('cross-owner read')
await other.close();await other_http.aclose();await c.close();await http.aclose();store.db.close()
print('persisted resume, completed/canceled ordering and owner checks passed')
asyncio.run(main())发送第一条消息后得到task/context，关闭客户端和数据库连接，由新Python进程读取持久快照，再重建整个服务，Get Task仍取回同一身份。补参消息同时携带task_id与context_id，不能仅复用对话context新建任务。SQLite保存的是服务端Task快照；序列化快照不等于后台计算自动恢复，本例仅在下一次补参请求时执行。主动运行中的任务还要保存业务步骤与外部副作用账本。
取消与完成没有“先发先赢”的保证
实验先完成再取消，服务器返回TaskNotCancelableError，查询保持COMPLETED；另一个INPUT_REQUIRED任务先取消，查询成为CANCELED。它验证两种顺序的结果，不宣称覆盖真正并发调度的所有竞态。实际并发必须在任务账本用原子条件更新决定唯一终态，再禁止旧worker提交；取消请求的HTTP成功与副作用撤销也不是同一件事。
身份不能取自消息中的 tenant
本地示例将两个固定演示Bearer值映射为服务端owner，仅供本地测试，不能部署为认证方案。A2A请求的tenant是路由数据，SDK可以从请求体设置它；因此授权使用ContextBuilder验证后保存在state的owner，不使用模型/客户端可任意指定的tenant字段。生产入口应验证真实访问令牌的签名、受众和期限，再映射最小权限身份。跨owner的Get Task返回不存在，避免泄露任务内容。
产物是分块更新，不是每次覆盖
同一artifact_id='stock'先发送A:（append=False），再发送3（append=True，last_chunk=True），SDK聚合快照为A:3。客户端必须按task_id+artifact_id隔离缓冲区；换artifact_id是另一份产物，last_chunk结束该产物而不是替代任务COMPLETED。网络断连后的Get Task返回当前快照，不能假定重订阅一定补发所有中间块。
SQLite示例采用单进程连接，list不实现分页，仅用于小数据集。多worker部署需要条件更新、版本检查、连接管理与分页，不应把该教学store当可直接上线的数据库适配器。
练习
把第二个补参值改为sku:B，任务应继续INPUT_REQUIRED而不能生成A:3；对该任务取消后再次Get Task仍为CANCELED。把第二个artifact块的append改为False，预测结果并运行：它是替换而非追加。解释为什么客户端不能把两者都拼接。