紧接上文

五、最简 A2A 应用实现过程

一个极简但功能完整的入门案例（包括代码、环境安装、运行调试过程说明等），是帮助读者快速了解 A2A 应用开发的重要实践方式。

5.1 源代码结构详解

根据 A2A 的架构设计，源代码分为 A2A Server 和 A2A Client 两个部分。

1. A2A Server源代码

time_agent_server.py实现了一个对外提供当前时间查询服务的Agent，尽管代码只有50多行，却涵盖了A2A规范的大部分核心内容。

1）引入依赖库

• import uvicorn
  from datetime import datetime
  from a2a.server.apps import A2AStarletteApplication
  from a2a.server.request_handlers import DefaultRequestHandler
  from a2a.server.agent_execution import AgentExecutor, RequestContext
  from a2a.server.tasks import InMemoryTaskStore
  from a2a.server.events import EventQueue
  from a2a.types import AgentCapabilities, AgentCard, AgentSkill
  from a2a.utils import new_agent_text_message

各依赖库或类的用途说明如下表所示。

2）定义 Agent Card

• skill = AgentSkill(
   id='current-time-skill',
   name='当前时间查询',
   description='获取当前的系统时间',
   tags=['时间服务', '实时查询', '工具类'],
   examples=['现在几点了？', '当前时间是多少？'],
  )
  public_agent_card = AgentCard(
   name='时间服务智能体',
   description='时间服务智能体，提供时间相关服务',
   url='http://localhost:9998/',
   version='1.0.0',
   default_input_modes=['text'],
   default_output_modes=['text'],
   capabilities=AgentCapabilities(streaming=True),
   skills=[skill],
   supports_authenticated_extended_card=True,
  )

Agent的业务能力通过AgentSkill实例进行描述，而该实例是Agent Card的一个属性。Capabilities字段中设置的“streaming=True”表示A2A Server支持流式传输。程序运行后，Agent Card的相关信息在程序运行时会在URLhttp://localhost:9998/.well-known/agent-card.json中向客户端公开。

3）实现 Agent

• class TimeAgent:
   async def invoke(self) -> str:
   return str(datetime.now())

TimeAgent提供了一个名为invoke的函数，以异步方式返回系统当前的时间。这部分代码即为本案例中最核心、最直接的业务逻辑实现。

4）实现 AgentExecutor

• class TimeAgentExecutor(AgentExecutor):
   def __init__(self):
   self.agent = TimeAgent()
   async def execute(
   self,
   context: RequestContext,
   event_queue: EventQueue,
   ) -> None:
   text = context.get_user_input()
   print("客户端输入：" + text)
   result = await self.agent.invoke()
   await event_queue.enqueue_event(
   new_agent_text_message(result))
   async def cancel(
   self, context: RequestContext, event_queue: EventQueue
   ) -> None:
   raise Exception('cancel not supported')

TimeAgentExecutor 是 TimeAgent 的执行器，用于调度 Agent 并向客户端推送消息。作为AgentExecutor 的子类，它必须实现 execute 和 cancel 两个方法。在 execute 方法中，执行流程如下：

（1）通过RequestContext获取客户端传入的文本信息。

（2）调用TimeAgent.invoke()方法获取当前时间。

（3）使用new_agent_text_message工具函数将结果封装成符合A2A规范的消息格式。

（4）通过EventQueue将消息加入事件队列，以便异步推送给客户端。

cancel方法在本示例中暂未实现具体逻辑，而是直接抛出“不支持取消操作”的异常。实际应用中可根据业务需求补充状态清理、任务终止等处理逻辑。当客户端主动调用cancel_task时，cancel方法会被触发执行。

execute方法中的几个重点解读如下：

（1）客户端传过来的消息通过context.get_user_input()获取。通过context还可以得到context_Id等代表消息上下文的信息，这在客户端被多次调用、需要利用历史会话记录的场景中非常有用。

（2）TimeAgent的invoke是异步方法，这里通过await简单地将其作为同步调用来获取返回值。

事实上，在大部分Agent的实现中，由于需要与大语言模型交互，通常会以逐步生成增量内容的异步调用方式实现。在execute中，往往也需要采取异步的方式接收Agent的返回结果，再将其推送给客户端。

（3）与常规的HTTP请求应答模式不同，A2A在客户端和服务端之间采用异步方式传递消息，机制上是基于事件驱动。本例中的体现是：将生成的文本消息推送到事件队列中，供后续异步传输给客户端。

5）RequestHandler

• request_handler = DefaultRequestHandler(
   agent_executor=TimeAgentExecutor(),
   task_store=InMemoryTaskStore(),
  )

DefaultRequestHandler的实例负责将客户端的请求分发给AgentExecutor去处理。为了管理任务的上下文信息，这里使用InMemoryTaskStore在内存中存储任务执行过程中的状态数据。

6）启动 A2A Server

• server = A2AStarletteApplication(
   agent_card=public_agent_card,
   http_handler=request_handler
  )
  uvicorn.run(server.build(), host='0.0.0.0', port=9998)

这里定义了一个支持JSON-RPC 2.0通道的A2AStarletteApplication，并通过uvicorn启动服务，监听本机的9998端口以接收客户端的请求。

2. A2A Client源代码

time_agent_client.py使用极简的代码调用A2A Server，总共20多行代码。

1）引入依赖库

• import asyncio
  from uuid import uuid4
  from a2a.client import (
   ClientConfig, ClientFactory, minimal_agent_card)
  from a2a.types import Message, Role, TextPart
  from a2a.utils.message import get_message_text

依赖库或类的用途说明如下表所示。

2）调用 A2A Server

• async def main():
   _client_factory = ClientFactory(ClientConfig())
   client = _client_factory.create(
   minimal_agent_card('http://localhost:9998')
   )
   msg = Message(
   kind='message',
   message_id=uuid4().hex,
   role=Role.user,
   parts=[TextPart(text='你好')]
   )
   async for response in client.send_message(msg):
  print(get_message_text(response))
  if __name__ == "__main__":
   asyncio.run(main())

首先通过minimal_agent_card构造（或获取）A2A Server公布的Agent Card信息，并使用ClientFactory 创 建 A2A Client 对 象 ； 然 后 构 造 一 个 包 含 TextPart 的 Message 对象， 通 过client.send_message(msg)发起调用，以异步迭代的方式处理流式响应，并解析返回的消息。最后的主函数为标准的异步程序启动方式，使用asyncio运行主协程。

5.2 运行最简 A2A 应用

首先安装运行环境，然后分别在不同的命令行窗口运行 A2A Server 和 A2A Client，观察应用两端的运行情况。

1. 环境安装

创建一个库依赖关系文件requirements.txt，内容如下：

• a2a-sdk[http-server]==0.3.10
  uvicorn==0.38.0

然后运行以下命令安装程序的运行环境。

• # 创建虚拟环境
  conda create -n chapter02 python=3.13 -y
  # 激活虚拟环境
  conda activate chapter02
  # 安装依赖库
  pip install -r requirements.txt -i https://pypi.mirrors.ustc.edu.cn/simple

2. 运行A2A Server

打开一个命令行窗口，运行以下命令：

• conda activate chapter02
  python time_agent_server.py

运行结果如下图所示。

在浏览器中访问“http://127.0.0.1:9998/.well-known/agent-card.json”，可以查看A2A Server公布的Agent Card信息，结果如下：

• {
   "capabilities": {
   "streaming": true
   },
   "defaultInputModes": [
   "text"
   ],
   "defaultOutputModes": [
   "text"
   ],
   "description": "时间服务智能体，提供时间相关服务",
   "name": "时间服务智能体",
   "preferredTransport": "JSONRPC",
   "protocolVersion": "0.3.0",
   "skills": [
   {
   "description": "获取当前的系统时间",
   "examples": [
   "现在几点了？",
   "当前时间是多少？"
   ],
   "id": "current-time-skill",
   "name": "当前时间查询",
   "tags": [
   "时间服务",
   "实时查询",
   "工具类"
   ]
   }
   ],
   "supportsAuthenticatedExtendedCard": true,
   "url": "http://localhost:9998/",
   "version": "1.0.0"
  }
  

3. 运行A2A Client

另开一个命令行窗口，运行以下命令：

• conda activate chapter02
  python time_agent_client.py

运行结果为输出当前时间。

六、本文小结

本文主要讲解了开发环境的搭建与一个简单示例，目的是帮助读者快速熟悉A2A在实际开发中的应用流程。本文从环境搭建入手，以获取当前时间的TimeAgent应用为例，详细解读了A2A应用的架构，并对A2A Server和A2A Client的实现进行了详细解读，为读者基于A2A开发更复杂的Agent应用奠定了基础。后续章节将在此基础上深入探讨更多高级特性和实际业务场景的应用。

本文摘自《构建自主AI:深入A2A协议的智能体开发》，具体内容请以书籍为准。

构建自主AI：深入A2A协议的智能体开发——jd