Running ADK Java Tools Asynchronously with CompletableFuture, in depth: bridging futures to Single, executors, fan-out, timeouts, cancellation and tests

What FunctionTool does with your return value

FunctionTool wraps a Java method found by reflection, static or on an instance you pass to FunctionTool.create(instance, "methodName"). When the model calls the tool, runAsync(args, toolContext) builds the argument array and invokes your method straight away, inside that call, before anything subscribes. It then inspects the returned object. A Maybe is filtered and mapped to a result map. A Single is converted to a Maybe and treated the same way. Null or an empty Optional produces an empty result. Anything else is treated as a plain value and converted to a map with Jackson; if that conversion fails, it is wrapped as a map under the key result.

CompletableFuture is not on that list. A method that returns one is not awaited. FunctionTool passes the future object itself to the Jackson conversion, so whatever the model receives, it is never the value the future will eventually hold. ADK Java on main has no native CompletableFuture support: you bridge to Single yourself.

Two more consequences follow from the method being invoked inside runAsync. First, whatever your method does before it returns runs on the thread that called runAsync. Do the minimum there: start the future and return. Second, an exception thrown synchronously by the method is caught by runAsync, logged, and turned into a generic internal-error result. A failure that arrives later, through a failed Single, travels down the reactive chain as an error instead, and how far it propagates depends on your callbacks; see ADK Java error recovery. The safest contract is to never let the Single fail: convert every failure into a result map yourself.

// Looks async, is not usable: FunctionTool does not await CompletableFuture.
@Schema(description = "Get a shipping quote for an order")
public CompletableFuture<Map<String, Object>> getShippingQuote(
        @Schema(name = "orderId", description = "Order id") String orderId) {
    return client.quoteAsync(orderId).thenApply(q -> Map.of("price", q.price()));
}

// Looks fine, blocks the caller thread for the whole HTTP call.
public Map<String, Object> getShippingQuoteBlocking(String orderId) {
    return client.quoteAsync(orderId).thenApply(q -> Map.<String, Object>of("price", q.price())).join();
}

The bridge: Single.fromCompletionStage, deferred

import com.google.adk.tools.Annotations.Schema;
import io.reactivex.rxjava3.core.Single;
import java.util.Map;
import java.util.concurrent.*;

public final class ShippingTools {
    private final ShippingClient client;   // returns CompletableFuture<Quote>
    private final Executor io;             // bounded, owned by this class, not the common pool
    private final Semaphore permits = new Semaphore(20);   // vendor allows 20 in flight

    public ShippingTools(ShippingClient client, Executor io) {
        this.client = client;
        this.io = io;
    }

    @Schema(description = "Get a shipping quote for an order. Returns price and carrier, or status=error.")
    public Single<Map<String, Object>> getShippingQuote(
            @Schema(name = "orderId", description = "The order id, for example ORD-1042") String orderId) {
        return Single.defer(() -> Single.fromCompletionStage(quote(orderId)))
                .timeout(4, TimeUnit.SECONDS)
                .onErrorReturn(e -> errorResult(unwrap(e)));
    }

    private CompletableFuture<Map<String, Object>> quote(String orderId) {
        if (!permits.tryAcquire()) {
            return CompletableFuture.completedFuture(
                Map.of("status", "error", "message", "shipping service busy, try again shortly"));
        }
        return client.quoteAsync(orderId)                        // starts the HTTP call
                .thenApplyAsync(q -> Map.<String, Object>of(
                        "status", "ok", "price", q.price(), "carrier", q.carrier()), io)
                .whenComplete((r, e) -> permits.release());
    }

    static Throwable unwrap(Throwable e) {
        while ((e instanceof CompletionException || e instanceof ExecutionException) && e.getCause() != null) {
            e = e.getCause();
        }
        return e;
    }

    static Map<String, Object> errorResult(Throwable e) {
        String msg = (e instanceof TimeoutException) ? "shipping service timed out"
                   : "shipping service failed: " + e.getClass().getSimpleName();
        return Map.of("status", "error", "message", msg);
    }
}

// Registration
ShippingTools tools = new ShippingTools(client, ioExecutor);
FunctionTool quoteTool = FunctionTool.create(tools, "getShippingQuote");

The method returns Single<Map<String, Object>>, which FunctionTool recognises. Single.fromCompletionStage adapts a stage that already exists. Wrapping it in Single.defer moves the call that creates the future to subscription time, so a retry operator or a second subscriber starts a fresh request instead of replaying the first future's result. Without defer, retry() silently resubscribes to the same completed future and never calls the backend again.

Three semantics of the bridge matter. A stage that completes with null signals an error for Single, so never complete with null; if absence is meaningful, return a map that says so. Disposing the Single, for example when the timeout fires, detaches from the stage but does not cancel it, because CompletionStage has no cancellation; the RxJava documentation says this explicitly. And the Single emits on whichever thread completes the future, which is why the mapping step uses thenApplyAsync(..., io): you control the thread that runs your mapping code rather than inheriting an HTTP client's selector thread.

Choosing the executor

If you call CompletableFuture.supplyAsync(supplier) with no executor, the work runs on the ForkJoinPool common pool, which is sized to the number of CPU cores and shared with parallel streams and everything else in the JVM. A few slow blocking calls there starve unrelated code. Always pass an executor. The choice depends on what the work does.

Virtual threads remove the cost of a thread, not the backend's limit. Twenty concurrent model turns, each calling three tools, can open sixty connections to a vendor that allows twenty. The semaphore in the example is a bulkhead: when it is full the tool returns a busy result immediately instead of queueing, and the model can tell the user or try later. Size it from the backend's documented limit, and give each backend its own semaphore so one slow dependency cannot exhaust the others.

Parallel tool calls: what ADK does and what you must do

When one model response contains several function calls, the Functions flow on main maps them through concatMapEager: every tool's Single is subscribed eagerly and results are emitted in the order the model requested, whatever order they finish in. RunConfig carries a ToolExecutionMode with the values NONE (treated as PARALLEL), SEQUENTIAL, PARALLEL and PARALLEL_SUBSCRIBE; the last also subscribes each tool on a worker thread. The thread-model article covers when to pick each.

The part that is yours: eager subscription only produces overlap if each tool's Single completes asynchronously. Three tools that each call join() inside the method run one after another on the caller thread, because each blocks inside runAsync before the next is even invoked. PARALLEL_SUBSCRIBE can rescue blocking tools by moving each onto a worker thread, at the cost of a thread per call. Three tools that start futures and return Singles overlap fully, and the turn takes as long as the slowest tool instead of the sum. That is the whole performance case for async tools: in a turn with a 900 ms, a 700 ms and a 400 ms call, the blocking version waits about 2 seconds and the async version about 0.9 seconds before the model runs again.

Fan-out inside one tool

// One tool call, three backends in parallel, partial results allowed.
private CompletableFuture<Map<String, Object>> orderStatus(String orderId) {
    CompletableFuture<Order>    order    = orders.getAsync(orderId);
    CompletableFuture<Payment>  payment  = payments.getAsync(orderId)
            .completeOnTimeout(null, 1500, TimeUnit.MILLISECONDS);   // optional part
    CompletableFuture<Shipment> shipment = shipping.trackAsync(orderId)
            .exceptionally(e -> null);                               // optional part

    return CompletableFuture.allOf(order, payment, shipment).thenApply(v -> {
        Map<String, Object> out = new LinkedHashMap<>();
        out.put("status", "ok");
        out.put("order", order.join().summary());          // safe: allOf has completed
        out.put("payment", payment.join() == null ? "unavailable" : payment.join().state());
        out.put("shipment", shipment.join() == null ? "unavailable" : shipment.join().state());
        return out;
    });
}

Sometimes one tool needs several backends, and you would rather give the model one compact answer than make it call three tools. allOf waits for all the futures; calling join() afterwards is safe because the values are already there. Decide per backend whether it is required or optional. Here the order is required, so its failure fails the whole tool, while payment and shipment are optional: completeOnTimeout supplies a fallback after 1.5 seconds and exceptionally turns failure into absence. Use thenCompose when one call needs another's result, such as looking up a customer id and then fetching their orders,.

Timeouts, failures and cancellation

There is no tool timeout in the Functions flow on main, so a backend that never answers holds the turn open. Put a timeout on every async tool. Two options exist. orTimeout(4, SECONDS) on the future completes it exceptionally with TimeoutException. RxJava's .timeout(4, TimeUnit.SECONDS) on the Single emits TimeoutException and disposes the upstream. Neither stops the work. The HTTP request keeps running and its permit stays taken until it finishes, so release resources in whenComplete on the future, as the example does, not in a Rx operator that never runs after disposal.

If you need the backend call actually aborted, keep a reference to the future and cancel it from doOnDispose. Cancelling a plain CompletableFuture does not interrupt the thread computing it; whether the underlying operation stops depends on the client, so check its documentation before you rely on it.

Then turn errors into results. Exceptions that pass through dependent stages arrive wrapped in CompletionException, which is why unwrap exists: classify the real cause and return status and message fields that the model can act on, such as timed out, not found or busy. Never put stack traces or hostnames in the message.

Context propagation

Work that hops threads loses thread-local context: the logging MDC, OpenTelemetry's current span, security principals. ADK's own tool spans are created around the call (see execute_tool in tool observability and metrics), but your backend client's spans will float free unless the context crosses the executor. With OpenTelemetry, wrap the executor once using Context.taskWrapping(executor), so every task carries the context it was submitted from. For MDC, copy the map at submission time and restore it inside the task.

Long-running work

Some operations take minutes: a report build, a human approval, a batch job. Do not hold a tool call open for those. Return immediately with a job id and a pending status, and let the result come back later as a function response the client submits. The main branch exposes this as an isLongRunning flag on FunctionTool.create(instance, methodName, requireConfirmation, isLongRunning), and the Functions flow skips the response event for a long-running tool that returns no result. ADK Java also has a LongRunningFunctionTool class; check which form your version offers. The future in this case belongs to your job system, not to the tool call.

Testing without sleeps

@Test
void quoteCompletesWhenFutureCompletes() {
    CompletableFuture<Quote> pending = new CompletableFuture<>();
    when(client.quoteAsync("ORD-1")).thenReturn(pending);
    ShippingTools tools = new ShippingTools(client, Runnable::run);   // same-thread executor

    TestObserver<Map<String, Object>> obs = tools.getShippingQuote("ORD-1").test();
    obs.assertNotComplete();                                          // nothing yet

    pending.complete(new Quote(12.5, "DHL"));
    obs.assertValue(m -> "ok".equals(m.get("status")) && m.get("price").equals(12.5));
}

@Test
void failureBecomesErrorResultNotException() {
    when(client.quoteAsync("ORD-2"))
        .thenReturn(CompletableFuture.failedFuture(new IOException("reset")));
    new ShippingTools(client, Runnable::run).getShippingQuote("ORD-2").test()
        .assertValue(m -> "error".equals(m.get("status")));
}

Hand-completed futures make async tests deterministic. Stub the client to return a CompletableFuture you hold, subscribe with test(), assert that nothing has been emitted, complete the future, then assert the result. A same-thread executor (Runnable::run) keeps everything on the test thread. For timeouts, pass an RxJava TestScheduler into the timeout operator and advance it manually, rather than waiting four real seconds. Add one test that the method returns quickly while the future is still pending; it catches the accidental join that kills parallelism.

Trade-offs

Async tools cost complexity: two concurrency models, late errors on other threads, and careful tests. They pay off when a turn calls several I/O-bound tools or tools wait on slow backends. For a single fast, local tool, a plain synchronous method returning a Map is simpler, and FunctionTool adapts it for you, as sync versus async contracts in ADK Java describes. Keep the async style for tools that wait.

What to do next

1. Search your tools for methods returning CompletableFuture and change them to return Single via Single.defer and fromCompletionStage.
2. Search for join() and get() inside tool methods and remove them from every I/O-bound tool.
3. Give every supplyAsync an explicit executor and add a semaphore bulkhead per backend sized from its limits.
4. Add a timeout and an onErrorReturn that produces status and message fields to every async tool.
5. Wrap your tool executors with OpenTelemetry context propagation and confirm backend spans nest under execute_tool.
6. Write a hand-completed-future test and a returns-quickly test for each async tool.
7. Confirm which ToolExecutionMode and long-running API your ADK version actually has before relying on either.