[Feature] Parallel Tool Call Execution

api/src/main/java/org/apache/flink/agents/api/agents/AgentExecutionOptions.java

@@ -20,6 +20,8 @@
 
 import org.apache.flink.agents.api.configuration.ConfigOption;
 
+import java.time.Duration;
+
 public class AgentExecutionOptions {
     public static final ConfigOption<Agent.ErrorHandlingStrategy> ERROR_HANDLING_STRATEGY =
             new ConfigOption<>(
@@ -42,9 +44,43 @@ public class AgentExecutionOptions {
     public static final ConfigOption<Boolean> CHAT_ASYNC =
             new ConfigOption<>("chat.async", Boolean.class, true);
 
+    /** Whether the built-in tool-call action runs each tool via durable async execution. */
     public static final ConfigOption<Boolean> TOOL_CALL_ASYNC =
             new ConfigOption<>("tool-call.async", Boolean.class, true);
 
+    /**
+     * Whether multiple tool calls from one {@code ToolRequestEvent} run as one parallel durable
+     * batch when {@link #TOOL_CALL_ASYNC} is also enabled (JDK 21+).
+     *
+     * <p>Default is {@code true}. A parallel batch raises the number of in-flight external calls;
+     * after failover, tools whose results were not yet persisted may be submitted again.
+     * Side-effecting tools should be idempotent or provide a {@code reconciler()}. Set to {@code
+     * false} to keep serial async or sync tool execution.
+     */
+    public static final ConfigOption<Boolean> TOOL_CALL_PARALLEL =
+            new ConfigOption<>("tool-call.parallel", Boolean.class, true);
+
+    /**
+     * Size of the dedicated thread pool used for tool-call async and parallel batch execution.
+     *
+     * <p>Separate from {@link #NUM_ASYNC_THREADS} so a large tool batch does not exhaust the global
+     * async pool.
+     */
+    public static final ConfigOption<Integer> TOOL_CALL_NUM_ASYNC_THREADS =
+            new ConfigOption<>(
+                    "tool-call.num-async-threads",
+                    Integer.class,
+                    Runtime.getRuntime().availableProcessors() * 2);
+
+    /**
+     * Overall timeout for one parallel tool-call batch.
+     *
+     * <p>Non-positive values disable the timeout. When the deadline elapses, unfinished slots are
+     * failed; slots that already completed keep their success or failure outcome.
+     */
+    public static final ConfigOption<Duration> TOOL_CALL_BATCH_TIMEOUT =
+            new ConfigOption<>("tool-call.batch.timeout", Duration.class, Duration.ofMillis(-1));
+
     public static final ConfigOption<Boolean> RAG_ASYNC =
             new ConfigOption<>("rag.async", Boolean.class, true);
 

api/src/main/java/org/apache/flink/agents/api/context/Outcome.java

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.api.context;
+
+/** Result of a durable batch execution slot. */
+public class Outcome<T> {
+    private final T value;
+    private final Exception error;
+
+    private Outcome(T value, Exception error) {
+        this.value = value;
+        this.error = error;
+    }
+
+    public static <T> Outcome<T> success(T value) {
+        return new Outcome<>(value, null);
+    }
+
+    public static <T> Outcome<T> failure(Exception error) {
+        return new Outcome<>(null, error);
+    }
+
+    public boolean isSuccess() {
+        return error == null;
+    }
+
+    public boolean isFailure() {
+        return error != null;
+    }
+
+    public T getValue() {
+        return value;
+    }
+
+    public Exception getError() {
+        return error;
+    }
+}

api/src/main/java/org/apache/flink/agents/api/context/RunnerContext.java

@@ -24,6 +24,7 @@
 import org.apache.flink.agents.api.resource.Resource;
 import org.apache.flink.agents.api.resource.ResourceType;
 
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -146,6 +147,21 @@ public interface RunnerContext {
      */
     <T> T durableExecuteAsync(DurableCallable<T> callable) throws Exception;
 
+    /**
+     * Executes multiple durable callables as one batch and returns outcomes in input order.
+     *
+     * <p>On JDK 21+, implementations may submit uncached calls concurrently and yield the current
+     * action execution until the batch completes. On JDK &lt; 21, this falls back to serial durable
+     * execution.
+     *
+     * <p>The callable list must be deterministic across recovery: same order and same {@link
+     * DurableCallable#getId()} values.
+     *
+     * <p>Access to memory and sendEvent are prohibited within the callables.
+     */
+    <T> List<Outcome<T>> durableExecuteAllAsync(List<DurableCallable<T>> callables)
+            throws Exception;
+
     /** Clean up the resource. */
     void close() throws Exception;
 }

docs/content/docs/operations/configuration.md

@@ -131,7 +131,10 @@ Here is the list of all built-in core configuration options.
 | `max-retries`             | 3                          | int                   | Number of retries when using `ErrorHandlingStrategy.RETRY`.                                                                                                                                                                                                     |
 | `retry-wait-interval`     | 1                          | int                   | Base wait interval in seconds between retries when using `ErrorHandlingStrategy.RETRY`. Uses exponential backoff: the actual wait time for the Nth retry is `retry-wait-interval * 2^(N-1)` seconds. For example, with default 1s, waits are 1s, 2s, 4s, etc. Retry count and total wait time are reported in `ChatResponseEvent` and recorded as metrics (`retryCount`, `retryWaitSec`) under the connection name. |
 | `chat.async`              | true                       | boolean               | Whether chat asynchronously for built-in chat action.                                                                                                                                                                                                           |
-| `tool-call.async`         | true                       | boolean               | Whether process tool call for built-in tool call action.                                                                                                                                                                                                        |
+| `tool-call.async`         | true                       | boolean               | Whether the built-in tool-call action runs each tool via durable async execution.                                                                                                                                                                               |
+| `tool-call.parallel`      | true                       | boolean               | When `tool-call.async` is also true (JDK 21+), run multiple tool calls from one `ToolRequestEvent` as one parallel durable batch. Increases in-flight external calls; after failover, unfinished tools may be submitted again — side-effecting tools should be idempotent or provide a reconciler. Set to `false` for serial tool execution. |
+| `tool-call.num-async-threads` | os cpu count * 2       | int                   | Dedicated thread pool size for tool-call async / parallel batch execution. Separate from `num-async-threads` so a large tool batch does not exhaust the global async pool.                                                                                    |
+| `tool-call.batch.timeout` | -1ms (disabled)            | Duration              | Overall timeout for one parallel tool-call batch. Non-positive disables it. On timeout, completed slots keep their outcome and unfinished slots fail. Timeout cancellation is best-effort; external side effects from unfinished tool calls may still complete, so side-effecting tools should be idempotent or provide a reconciler. |
 | `rag.async`               | true                       | boolean               | Whether retrieve context asynchronously for built-in context retrieval action.                                                                                                                                                                                  |
 | `num-async-threads`       | os cpu count * 2           | int                   | The thread pool size for async executor.                                                                                                                                                                                                                        |
 | `job-identifier`          | none                       | String                | The unique identifier of job, remaining consistent after restoring from a savepoint. If not set, uses flink job id.                                                                                                                                             |

e2e-test/flink-agents-end-to-end-tests-integration/src/test/java/org/apache/flink/agents/integration/test/AsyncExecutionAgent.java

@@ -23,13 +23,33 @@
 import org.apache.flink.agents.api.OutputEvent;
 import org.apache.flink.agents.api.agents.Agent;
 import org.apache.flink.agents.api.annotation.Action;
+import org.apache.flink.agents.api.annotation.ChatModelConnection;
+import org.apache.flink.agents.api.annotation.ChatModelSetup;
+import org.apache.flink.agents.api.annotation.ToolParam;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.chat.model.BaseChatModelConnection;
+import org.apache.flink.agents.api.chat.model.BaseChatModelSetup;
 import org.apache.flink.agents.api.context.DurableCallable;
 import org.apache.flink.agents.api.context.MemoryObject;
 import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.tools.Tool;
+import org.apache.flink.agents.api.tools.ToolMetadata;
+import org.apache.flink.agents.api.tools.ToolParameters;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.tools.ToolType;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 /**
  * Agent definition for testing async execution functionality.
  *
@@ -71,6 +91,141 @@ public Integer getKey(AsyncRequest request) {
         }
     }
 
+    /** Chat connection that emits one tool request containing multiple slow tool calls. */
+    public static class ToolBatchChatConnection extends BaseChatModelConnection {
+        public ToolBatchChatConnection(
+                ResourceDescriptor descriptor, ResourceContext resourceContext) {
+            super(descriptor, resourceContext);
+        }
+
+        @Override
+        public ChatMessage chat(
+                List<ChatMessage> messages, List<Tool> tools, Map<String, Object> modelParams) {
+            ChatMessage lastMessage = messages.get(messages.size() - 1);
+            if (lastMessage.getRole() == MessageRole.TOOL) {
+                return new ChatMessage(MessageRole.ASSISTANT, lastMessage.getContent());
+            }
+
+            String requestId = lastMessage.getContent();
+            return new ChatMessage(
+                    MessageRole.ASSISTANT,
+                    "",
+                    List.of(
+                            toolCall("call-1", requestId, 1),
+                            toolCall("call-2", requestId, 2),
+                            toolCall("call-3", requestId, 3)));
+        }
+    }
+
+    /** Chat model setup bound to the timed tool. */
+    public static class ToolBatchChatModel extends BaseChatModelSetup {
+        public ToolBatchChatModel(ResourceDescriptor descriptor, ResourceContext resourceContext) {
+            super(descriptor, resourceContext);
+        }
+
+        @Override
+        public Map<String, Object> getParameters() {
+            return new HashMap<>();
+        }
+    }
+
+    private static Map<String, Object> toolCall(String id, String requestId, int index) {
+        return Map.of(
+                "id",
+                id,
+                "type",
+                "function",
+                "function",
+                Map.of(
+                        "name",
+                        "timed_tool",
+                        "arguments",
+                        Map.of("request_id", requestId, "call_index", index)));
+    }
+
+    /** Agent that requests one chat turn that produces multiple slow tool calls. */
+    public static class ToolBatchAgent extends Agent {
+        public ToolBatchAgent(int sleepTimeMs) {}
+
+        @ChatModelConnection
+        public static ResourceDescriptor toolBatchChatConnection() {
+            return ResourceDescriptor.Builder.newBuilder(ToolBatchChatConnection.class.getName())
+                    .build();
+        }
+
+        @ChatModelSetup
+        public static ResourceDescriptor toolBatchChatModel() {
+            return ResourceDescriptor.Builder.newBuilder(ToolBatchChatModel.class.getName())
+                    .addInitialArgument("connection", "toolBatchChatConnection")
+                    .addInitialArgument("model", "test-model")
+                    .addInitialArgument("tools", List.of("timed_tool"))
+                    .build();
+        }
+
+        @org.apache.flink.agents.api.annotation.Tool(
+                description = "Records timing for a slow tool call.")
+        public static String timed_tool(
+                @ToolParam(name = "request_id") String requestId,
+                @ToolParam(name = "call_index") String callIndex) {
+            long start = System.currentTimeMillis();
+            try {
+                Thread.sleep(500);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            long end = System.currentTimeMillis();
+            return String.format(
+                    "request=%s,call=%s,start=%d,end=%d", requestId, callIndex, start, end);
+        }
+
+        @Action(EventType.InputEvent)
+        public static void requestTools(Event event, RunnerContext ctx) {
+            InputEvent inputEvent = InputEvent.fromEvent(event);
+            AsyncRequest request = (AsyncRequest) inputEvent.getInput();
+            ctx.sendEvent(
+                    new ChatRequestEvent(
+                            "toolBatchChatModel",
+                            List.of(
+                                    new ChatMessage(
+                                            MessageRole.USER, String.valueOf(request.id)))));
+        }
+
+        @Action(EventType.ChatResponseEvent)
+        public static void emitToolTimings(Event event, RunnerContext ctx) {
+            ChatResponseEvent responseEvent = ChatResponseEvent.fromEvent(event);
+            ctx.sendEvent(new OutputEvent(responseEvent.getResponse().getContent()));
+        }
+    }
+
+    /** Tool that records its execution time range in the response result. */
+    public static class TimedTool extends Tool {
+        private final int sleepTimeMs;
+
+        public TimedTool(int sleepTimeMs) {
+            super(new ToolMetadata("timed_tool", "Records timing for a slow tool call.", "{}"));
+            this.sleepTimeMs = sleepTimeMs;
+        }
+
+        @Override
+        public ToolType getToolType() {
+            return ToolType.FUNCTION;
+        }
+
+        @Override
+        public ToolResponse call(ToolParameters parameters) {
+            int callIndex = parameters.getParameter("call_index", Integer.class);
+            long start = System.currentTimeMillis();
+            try {
+                Thread.sleep(sleepTimeMs);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            long end = System.currentTimeMillis();
+            return ToolResponse.success(
+                    String.format("call=%d,start=%d,end=%d", callIndex, start, end));
+        }
+    }
+
     /** Custom event type for internal agent communication. */
     public static class AsyncProcessedEvent extends Event {
         public static final String EVENT_TYPE = "AsyncProcessedEvent";