refactor: Let LlmAgent.tools() return an async type instead of blocking

PiperOrigin-RevId: 843824506

Author: google-genai-bot

core/src/main/java/com/google/adk/agents/LlmAgent.java

@@ -736,8 +736,8 @@ public IncludeContents includeContents() {
     return includeContents;
   }
 
-  public List<BaseTool> tools() {
-    return canonicalTools().toList().blockingGet();
+  public Single<List<BaseTool>> tools() {
+    return canonicalTools().toList();
   }
 
   public List<Object> toolsUnion() {

core/src/main/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessor.java

@@ -106,11 +106,16 @@ private Maybe<Event> assembleEvent(
       InvocationContext invocationContext,
       Collection<FunctionCall> functionCalls,
       Map<String, ToolConfirmation> toolConfirmations) {
-    ImmutableMap.Builder<String, BaseTool> toolsBuilder = ImmutableMap.builder();
+    Single<ImmutableMap<String, BaseTool>> toolsMapSingle;
     if (invocationContext.agent() instanceof LlmAgent llmAgent) {
-      for (BaseTool tool : llmAgent.tools()) {
-        toolsBuilder.put(tool.name(), tool);
-      }
+      toolsMapSingle =
+          llmAgent
+              .tools()
+              .map(
+                  toolList ->
+                      toolList.stream().collect(toImmutableMap(BaseTool::name, tool -> tool)));
+    } else {
+      toolsMapSingle = Single.just(ImmutableMap.of());
     }
 
     var functionCallEvent =
@@ -124,8 +129,10 @@ private Maybe<Event> assembleEvent(
                     .build())
             .build();
 
-    return Functions.handleFunctionCalls(
-        invocationContext, functionCallEvent, toolsBuilder.buildOrThrow(), toolConfirmations);
+    return toolsMapSingle.flatMapMaybe(
+        toolsMap ->
+            Functions.handleFunctionCalls(
+                invocationContext, functionCallEvent, toolsMap, toolConfirmations));
   }
 
   private ImmutableMap<String, ToolConfirmation> filterRequestConfirmationFunctionResponses(

core/src/main/java/com/google/adk/runner/Runner.java

@@ -50,8 +50,8 @@
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Maybe;
 import io.reactivex.rxjava3.core.Single;
-import java.lang.reflect.Parameter;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -616,35 +616,39 @@ public Flowable<Event> runLive(
     try {
       InvocationContext invocationContext =
           newInvocationContextForLive(session, Optional.of(liveRequestQueue), runConfig);
-      if (invocationContext.agent() instanceof LlmAgent) {
-        LlmAgent agent = (LlmAgent) invocationContext.agent();
-        for (BaseTool tool : agent.tools()) {
-          if (tool instanceof FunctionTool functionTool) {
-            for (Parameter parameter : functionTool.func().getParameters()) {
-              if (parameter.getType().equals(LiveRequestQueue.class)) {
-                invocationContext
-                    .activeStreamingTools()
-                    .put(functionTool.name(), new ActiveStreamingTool(new LiveRequestQueue()));
-              }
-            }
-          }
-        }
+
+      Single<InvocationContext> invocationContextSingle;
+      if (invocationContext.agent() instanceof LlmAgent agent) {
+        invocationContextSingle =
+            agent
+                .tools()
+                .map(
+                    tools -> {
+                      this.addActiveStreamingTools(invocationContext, tools);
+                      return invocationContext;
+                    });
+      } else {
+        invocationContextSingle = Single.just(invocationContext);
       }
-      return Telemetry.traceFlowable(
-          spanContext,
-          span,
-          () ->
-              invocationContext
-                  .agent()
-                  .runLive(invocationContext)
-                  .doOnNext(event -> this.sessionService.appendEvent(session, event))
-                  .onErrorResumeNext(
-                      throwable -> {
-                        span.setStatus(StatusCode.ERROR, "Error in runLive Flowable execution");
-                        span.recordException(throwable);
-                        span.end();
-                        return Flowable.error(throwable);
-                      }));
+
+      return invocationContextSingle.flatMapPublisher(
+          updatedInvocationContext ->
+              Telemetry.traceFlowable(
+                  spanContext,
+                  span,
+                  () ->
+                      updatedInvocationContext
+                          .agent()
+                          .runLive(updatedInvocationContext)
+                          .doOnNext(event -> this.sessionService.appendEvent(session, event))
+                          .onErrorResumeNext(
+                              throwable -> {
+                                span.setStatus(
+                                    StatusCode.ERROR, "Error in runLive Flowable execution");
+                                span.recordException(throwable);
+                                span.end();
+                                return Flowable.error(throwable);
+                              })));
     } catch (Throwable t) {
       span.setStatus(StatusCode.ERROR, "Error during runLive synchronous setup");
       span.recordException(t);
@@ -740,5 +744,22 @@ private BaseAgent findAgentToRun(Session session, BaseAgent rootAgent) {
     return rootAgent;
   }
 
+  private void addActiveStreamingTools(InvocationContext invocationContext, List<BaseTool> tools) {
+    tools.stream()
+        .filter(FunctionTool.class::isInstance)
+        .map(FunctionTool.class::cast)
+        .filter(this::hasLiveRequestQueueParameter)
+        .forEach(
+            tool ->
+                invocationContext
+                    .activeStreamingTools()
+                    .put(tool.name(), new ActiveStreamingTool(new LiveRequestQueue())));
+  }
+
+  private boolean hasLiveRequestQueueParameter(FunctionTool functionTool) {
+    return Arrays.stream(functionTool.func().getParameters())
+        .anyMatch(parameter -> parameter.getType().equals(LiveRequestQueue.class));
+  }
+
   // TODO: run statelessly
 }

core/src/test/java/com/google/adk/agents/ConfigAgentUtilsTest.java

@@ -258,8 +258,8 @@ public void fromConfig_withBuiltInTool_loadsTool() throws IOException, Configura
 
     assertThat(agent).isInstanceOf(LlmAgent.class);
     LlmAgent llmAgent = (LlmAgent) agent;
-    assertThat(llmAgent.tools()).hasSize(1);
-    assertThat(llmAgent.tools().get(0).name()).isEqualTo("google_search");
+    assertThat(llmAgent.tools().blockingGet()).hasSize(1);
+    assertThat(llmAgent.tools().blockingGet().get(0).name()).isEqualTo("google_search");
   }
 
   @Test
@@ -784,7 +784,7 @@ public void fromConfig_withIncludeContentsAndOtherFields_parsesAllFieldsCorrectl
     assertThat(llmAgent.outputKey()).hasValue("testOutput");
     assertThat(llmAgent.disallowTransferToParent()).isTrue();
     assertThat(llmAgent.disallowTransferToPeers()).isFalse();
-    assertThat(llmAgent.tools()).hasSize(1);
+    assertThat(llmAgent.tools().blockingGet()).hasSize(1);
     assertThat(llmAgent.model()).isPresent();
   }