a2a4j-examples

a2a4j example project

什么是 A2A 协议

A2A（Agent2Agent）协议 是由 Google Cloud 推出的一个开放协议，旨在促进不同 AI 代理之间的互操作性。其主要目标是允许这些代理在动态的、多代理的生态系统中进行有效的通信和协作，无论它们是由不同的供应商构建的还是使用不同的技术框架。

A2A协议如何工作？

1. 能力说明书（Agent Card）
   每个AI智能体需提供一份JSON格式的“说明书”，明确告知其他智能体：“我能做什么”（如分析数据、预订机票）、“需要什么输入格式”、“如何验证身份”。这相当于企业员工的名片，让协作方快速识别可用资源，其他智能体想合作就能很快找到它、理解它的能力，省去了大量沟通障碍。
2. 任务派发与追踪
   当一个智能体想委托另一个智能体去完成什么事情，就像对外发布一份“合作项目意向书”。对方同意接单后，双方会记录一个Task ID，追踪项目进度、交换资料，直到该Task完成为止。 假设用户让“旅行规划Agent”安排行程，该Agent可通过A2A向“机票预订Agent”发送任务请求，并实时接收状态更新（如“已找到航班，正在比价”）。任务支持即时完成或长达数天的复杂流程，且结果（如生成的行程表）会以标准化格式返回。
3. 跨模态通信
   协议支持文本、图片、音视频等多种数据类型。例如医疗场景中，影像分析Agent可直接将CT图像传递给诊断Agent，无需中间格式转换。
4. 安全验证机制
   所有通信默认加密，并通过OAuth等企业级认证，确保只有授权智能体可参与协作，防止数据泄露。

A2A 的参与者

A2A 协议有三个参与者：

• 用户（User）：使用代理系统完成任务的用户（人类或服务）
• 客户端（Client）：负责转发用户请求的任务
• 服务端（Server）：负责接收客户端的任务，开发者须处理任务请求调用三方LLM API，并响应给客户端

使用a2a4j

前置条件

Branches:

• jdk8 examples of jdk8, SpringBoot 2.7.18
• main examples of jdk17, SpringBoot 3.4.5

Features:

• support spring mvc, reactor, sse
• support servlet and sse
• support webflux and sse
• more a2a4j example project, please refer to a2a4j-examples jdk8 and a2a4j-examples main
• support more LLM, eg.LangChain4j
• support redis as task storage

agent/server配置

1. 引入maven依赖

   <dependencies>
       <dependency>
           <groupId>io.github.pheonixhkbxoic</groupId>
           <artifactId>a2a4j-agent-mvc-spring-boot-starter</artifactId>
           <version>2.0.2</version>
       </dependency>
       <!-- 或 use webflux -->
       <!--    <dependency>-->
       <!--        <groupId>io.github.pheonixhkbxoic</groupId>-->
       <!--        <artifactId>a2a4j-agent-webflux-spring-boot-starter</artifactId>-->
       <!--        <version>2.0.2</version>-->
       <!--    </dependency>-->
   </dependencies>

2. 配置AgentCard实例

   @Bean
   public AgentCard agentCard() {
       AgentCapabilities capabilities = new AgentCapabilities();
       AgentSkill skill = AgentSkill.builder()
               .id("echoAgent")
               .name("echo agent")
               .description("just echo user message")
               .tags(List.of("echo"))
               .examples(Collections.singletonList("I'm big strong!"))
               .inputModes(Collections.singletonList("text"))
               .outputModes(Collections.singletonList("text"))
               .build();
       AgentCard agentCard = new AgentCard();
       agentCard.setName("echoAgent");
       agentCard.setDescription("echo agent, Answer the user's questions exactly as they are");
       agentCard.setUrl("http://127.0.0.1:" + port);
       agentCard.setVersion("2.0.2");
       agentCard.setCapabilities(capabilities);
       agentCard.setSkills(Collections.singletonList(skill));
       return agentCard;
   }

   AgentCard用来描述当前Agent Server所具有的能力，客户端启动时会连接到 http://{your_server_domain}/.well-known/agent.json 来获取AgentCard

3. 实现自定义AgentInvoker

   @Component
   public class EchoAgentInvoker implements AgentInvoker {
       @Resource
       private EchoAgent agent;
   
       @Override
       public Mono<List<Artifact>> invoke(SendTaskRequest request) {
           String userQuery = this.extractUserQuery(request.getParams());
           return agent.chat(userQuery)
                   .map(text -> {
                       Artifact artifact = Artifact.builder().name("answer").parts(List.of(new TextPart(text))).build();
                       return List.of(artifact);
                   });
       }
   
       @Override
       public Flux<StreamData> invokeStream(SendTaskStreamingRequest request) {
           String userQuery = this.extractUserQuery(request.getParams());
           return agent.chatStream(userQuery)
                   .map(text -> {
                       Message message = Message.builder().role(Role.AGENT).parts(List.of(new TextPart(text))).build();
                       return StreamData.builder().state(TaskState.WORKING).message(message).endStream(false).build();
                   })
                   .concatWithValues(StreamData.builder()
                           .state(TaskState.COMPLETED)
                           .message(Message.builder().role(Role.AGENT).parts(List.of(new TextPart(""))).build())
                           .endStream(true)
                           .build());
       }
   }

   注意事项：

   • 需要实现AgentInvoker接口，处理agent的调用与转换

4. 代码参考
   a2a4j-examples agents/echo-agent

host/client配置

1. 引入maven依赖

   <dependency>
       <groupId>io.github.pheonixhkbxoic</groupId>
       <artifactId>a2a4j-host-spring-boot-starter</artifactId>
       <version>2.0.2</version>
   </dependency>

2. 在配置文件(如application.xml)中配置相关属性

   a2a4j:
     host:
       # can be null
       notification:
         url: http://127.0.0.1:8989/notify
   
       agents:
         echoAgent:
           baseUrl: http://127.0.0.1:8901
   

   • 必须配置Agent Server的baseUrl，可配置多个
   • 选择性的配置Notification Server的baseUrl

3. 发送任务请求并处理响应

   @Slf4j
   @RestController
   public class AgentController {
       @Resource
       private A2AClientSet clientSet;
   
       @GetMapping("/chat")
       public ResponseEntity<Object> chat(String userId, String sessionId, String prompts) {
           A2AClient client = clientSet.getByConfigKey("echoAgent");
           TaskSendParams params = TaskSendParams.builder()
                   .id(Uuid.uuid4hex())
                   .sessionId(sessionId)
                   .historyLength(3)
                   .acceptedOutputModes(Collections.singletonList("text"))
                   .message(new Message(Role.USER, Collections.singletonList(new TextPart(prompts)), null))
                   .pushNotification(client.getPushNotificationConfig())
                   .build();
           log.info("chat params: {}", Util.toJson(params));
           try {
               String answer = client.sendTask(params)
                       .flatMap(sendTaskResponse -> {
                           if (sendTaskResponse.getError() != null) {
                               return Mono.error(new ValueError(Util.toJson(sendTaskResponse.getError())));
                           }
                           Task task = sendTaskResponse.getResult();
                           return Mono.just(task.getArtifacts().stream()
                                   .flatMap(t -> t.getParts().stream())
                                   .filter(p -> Part.TEXT.equals(p.getType()))
                                   .map(p -> ((TextPart) p).getText())
                                   .filter(t -> !Util.isEmpty(t))
                                   .collect(Collectors.joining("\n")));
                       })
                       .block();
               return ResponseEntity.ok(answer);
           } catch (A2AClientHTTPError e) {
               log.error("chat exception: {}", e.getMessage(), e);
               return ResponseEntity.status(e.getStatusCode()).body(e.getMessage());
           } catch (Exception e) {
               log.error("chat exception: {}", e.getMessage(), e);
               return ResponseEntity.internalServerError().body(e.getMessage());
           }
       }
   }

4. 代码参考
   a2a4j-examples hosts/standalone

notification server配置

1. 引入maven依赖

   <dependencies>
   
       <dependency>
           <groupId>io.github.pheonixhkbxoic</groupId>
           <artifactId>a2a4j-notification-mvc-spring-boot-starter</artifactId>
           <version>2.0.2</version>
       </dependency>
       <!-- 或 use webflux -->
       <!--    <dependency>-->
       <!--        <groupId>io.github.pheonixhkbxoic</groupId>-->
       <!--        <artifactId>a2a4j-notification-webflux-spring-boot-starter</artifactId>-->
       <!--        <version>2.0.2</version>-->
       <!--    </dependency>-->
   </dependencies>

2. 在配置文件(如application.xml)中配置相关属性

   a2a4j:
     notification:
       # default 
       endpoint: "/notify"
       jwksUrls:
         - http://127.0.0.1:8901/.well-known/jwks.json
   

   • 必须配置jwksUrls，可配置多个
   • 选择性的配置endpoint, 不配置时默认监听/notify, AgentServer中的配置a2a4j.host.notification:

3. 自定义监听器并实例化

   @Component
   public class NotificationListener extends WebMvcNotificationAdapter {
       protected final ScheduledThreadPoolExecutor scheduler;
   
       public NotificationListener(@Autowired A2a4jNotificationProperties a2a4jNotificationProperties) {
           super(a2a4jNotificationProperties.getEndpoint(), a2a4jNotificationProperties.getJwksUrls());
   
           // auto reloadJwks when Agent restart
           scheduler = new ScheduledThreadPoolExecutor(1);
           scheduler.scheduleAtFixedRate(() -> {
               if (verifyFailCount.get() != 0) {
                   this.reloadJwks();
               }
           }, 5, 5, TimeUnit.SECONDS);
       }
       // TODO 实现方法来处理通知，可以使用默认实现
   }

   注意：

   • 需要继承WebMvcNotificationAdapter类
   • 注入配置属性@Autowired A2a4jNotificationProperties a2a4jNotificationProperties 并通过super(a2a4jNotificationProperties.getEndpoint(), a2a4jNotificationProperties.getJwksUrls());实例化 PushNotificationReceiverAuth和监听指定的地址

4. 代码参考
   a2a4j-examples notification-listener

manual

If you just want to use a2a client manually.

1. 引入maven依赖

   <dependency>
       <groupId>io.github.pheonixhkbxoic</groupId>
       <artifactId>a2a4j-core</artifactId>
       <version>2.0.2</version>
   </dependency>

2. 使用A2AClient连接到Agent Server
   The agent server must has start

   String agentCardBaseUrl = "http://127.0.0.1:8901";
   AgentCard agentCard = new AgentCardResolver(agentCardBaseUrl).resolve();
   A2AClient client = new A2AClient(agentCard);
   
   private static void chat(A2AClient client, String prompts) {
       TaskSendParams params = TaskSendParams.builder()
               .id(Uuid.uuid4hex())
               .sessionId(Uuid.uuid4hex())
               .historyLength(3)
               .acceptedOutputModes(Collections.singletonList("text"))
               .message(new Message(Role.USER, Collections.singletonList(new TextPart(prompts)), null))
               .pushNotification(client.getPushNotificationConfig())
               .build();
       try {
           client.sendTask(params)
                   .flatMap(sendTaskResponse -> {
                       if (sendTaskResponse.getError() != null) {
                           return Mono.error(new ValueError(Util.toJson(sendTaskResponse.getError())));
                       }
                       Task task = sendTaskResponse.getResult();
                       return Mono.just(task.getArtifacts().stream()
                               .flatMap(t -> t.getParts().stream())
                               .filter(p -> Part.TEXT.equals(p.getType()))
                               .map(p -> ((TextPart) p).getText())
                               .filter(t -> !Util.isEmpty(t))
                               .collect(Collectors.joining("\n")));
                   })
                   .subscribe(System.out::println);
       } catch (Exception e) {
           System.out.println(e.getMessage());
       }
   }

3. 代码参考
   a2a4j-examples hosts cli

hosts

• cli
• standalone
• multiagent one router agent, multi other agent
• agent-router-adk
  one router agent, multi other agent with adk-java
  

TaskStore

AgentInvoker used by DefaultTaskManager in agent, in factor TaskManager storages Task and PushNotificationInfo by TaskStore
InMemoryTaskStore is default, you can use a2a4j-storage-redis-spring-boot-starter dependency if you want use redis to store tasks

<dependency>
    <groupId>io.github.pheonixhkbxoic</groupId>
    <artifactId>a2a4j-storage-redis-spring-boot-starter</artifactId>
    <version>2.0.2</version>
</dependency>

Star History