Agent Card 与真实本地服务

远程 Agent 暴露能力，不暴露内部推理

MCP 主要让应用取得工具与上下文；A2A 让应用向另一个 Agent 委托任务。远端如何规划和使用模型可以隐藏，但任务身份、状态和交付物必须清楚。Agent Card 描述技能、协议接口、输入输出媒体类型与能力；它是发现数据，不是授权凭证。

主实验使用 Python 3.12 与 a2a-sdk==1.1.0。它使用受规范定义的模型与真实 HTTP 服务，不需要调用 LLM。空目录执行：

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install 'a2a-sdk==1.1.0' uvicorn starlette httpx sse-starlette
python -m pip freeze > requirements.lock

保存 server.py：

import uvicorn
from starlette.applications import Starlette
from a2a.server.agent_execution import AgentExecutor,RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import InMemoryTaskStore,TaskUpdater
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes,create_jsonrpc_routes
from a2a.helpers import get_message_text,new_task_from_user_message,new_text_part
from a2a.types import AgentCard,AgentCapabilities,AgentInterface,AgentSkill,TaskState
class Echo(AgentExecutor):
    async def execute(self,context:RequestContext,event_queue:EventQueue):
        task=context.current_task or new_task_from_user_message(context.message)
        if not context.current_task: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)
        text=get_message_text(context.message)
        await updater.add_artifact(parts=[new_text_part(text='收到：'+text,media_type='text/plain')])
        await updater.update_status(state=TaskState.TASK_STATE_COMPLETED)
    async def cancel(self,context:RequestContext,event_queue:EventQueue):
        raise NotImplementedError('This short echo task does not support cancellation')
card=AgentCard(name='Echo',description='只回显输入的协议实验',version='1.0.0',
    default_input_modes=['text/plain'],default_output_modes=['text/plain'],
    capabilities=AgentCapabilities(streaming=True),
    supported_interfaces=[AgentInterface(protocol_binding='JSONRPC',
        url='http://127.0.0.1:9999',protocol_version='1.0')],
    skills=[AgentSkill(id='echo',name='Echo',description='回显文本',tags=['echo'])])
handler=DefaultRequestHandler(agent_executor=Echo(),task_store=InMemoryTaskStore(),agent_card=card)
app=Starlette(routes=create_agent_card_routes(card)+create_jsonrpc_routes(handler,'/'))
if __name__=='__main__':uvicorn.run(app,host='127.0.0.1',port=9999)

运行 python server.py 保持终端开启，再在另一终端激活环境，保存 client.py：

import asyncio,httpx
from a2a.client import A2ACardResolver,ClientConfig,create_client
from a2a.helpers import new_text_message
from a2a.types import Role,SendMessageRequest
async def main():
    async with httpx.AsyncClient() as http:
        card=await A2ACardResolver(httpx_client=http,base_url='http://127.0.0.1:9999').get_agent_card()
    print(card.name,card.supported_interfaces)
    client=await create_client(agent=card,client_config=ClientConfig(streaming=False))
    try:
        request=SendMessageRequest(message=new_text_message('你好',role=Role.ROLE_USER))
        async for result in client.send_message(request):print(result)
    finally:await client.close()
asyncio.run(main())

python client.py 应看到 Echo 的发现信息、包含“收到：你好”的产物与完成状态。它只证明协议互通，不证明 Agent 推理能力。InMemoryTaskStore 在进程重启后丢失状态，不适合作为持久任务账本。验证后 Ctrl+C 停止服务。

参阅 官方 Hello World 与 规范。

练习

修改客户端文本为“库存 A”，保持服务不变。预期产物应包含“收到：库存 A”，但不能真的查询库存。解释为什么技能描述不是额外业务实现。

先记录自己的输出与解释，再核对本章答案。