feat: support parallel tool calling

docs/wayflowcore/source/core/changelog.rst

@@ -7,6 +7,13 @@ WayFlow |current_version|
 Improvements
 ^^^^^^^^^^^^
 
+* **Parallel tool calling**
+
+  Added the ``parallel_tool_calls`` LLM setting. Independent server-side tool
+  calls in one response can now execute concurrently, while client-side and
+  confirmation-requiring tools remain sequential. The documentation includes
+  guidance for tools with side effects or shared state.
+
 * **Configurable additional object properties**
 
   ``ObjectProperty`` now supports JSON Schema ``additionalProperties`` configuration.

docs/wayflowcore/source/core/howtoguides/howto_build_assistants_with_tools.rst

@@ -266,6 +266,50 @@ The only difference is that the file path is provided as a conversation message
     :start-after: # .. start-##_Creating_and_running_an_agent_with_a_client_tool
     :end-before: # .. end-##_Creating_and_running_an_agent_with_a_client_tool
 
+
+Parallel tool calling
+=====================
+
+When an LLM returns several tool requests in one response, WayFlow can execute
+the requests concurrently. Parallel tool calling is enabled by default and is
+configured on the LLM with the ``parallel_tool_calls`` flag:
+
+.. code-block:: python
+
+    llm = OpenAIModel(
+        model_id="gpt-4o",
+        generation_config=LlmGenerationConfig(temperature=0),
+        parallel_tool_calls=True,
+    )
+
+The flag controls WayFlow's tool-execution behavior; it does not make a model
+emit multiple tool requests. The model must support and choose to return a
+batch of tool requests. If parallel tool calling is disabled, requests in a
+batch are executed sequentially.
+
+WayFlow executes a batch in parallel only when all of its requests are
+server-side tools. Batches containing a ``ClientTool`` are executed
+sequentially, because a client tool yields control to the client and may have
+side effects that must remain ordered. This also means that tool requests
+should not depend on another request in the same batch. If a server-side tool
+uses shared mutable state, a shared database connection, or has ordering or
+side-effect requirements, document those limitations in the tool description
+and disable parallel tool calling when necessary:
+
+.. code-block:: python
+
+    llm = OpenAIModel(
+        model_id="gpt-4o",
+        generation_config=LlmGenerationConfig(temperature=0),
+        parallel_tool_calls=False,
+    )
+
+Tool descriptions should make dependencies and side effects clear to the LLM.
+For example, describe whether a tool is read-only, whether it mutates shared
+state, and whether it must be called after another tool. When parallel tool
+calling is enabled, only independent tool requests should be emitted in the
+same batch.
+
 Agent Spec Exporting/Loading
 ============================
 

wayflowcore/src/wayflowcore/executors/_agentexecutor.py

@@ -4,6 +4,7 @@
 # (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
 # (UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
 
+import asyncio
 import logging
 import os
 import warnings
@@ -52,6 +53,7 @@
 from wayflowcore.planning import ExecutionPlan
 from wayflowcore.property import JsonSchemaParam, Property, StringProperty, _validate_strict_outputs
 from wayflowcore.tools import ClientTool, Tool, ToolRequest, ToolResult
+from wayflowcore.tools.servertools import ServerTool
 from wayflowcore.tools.tools import _descriptors_to_json_schema_map, _sanitize_tool_name
 from wayflowcore.tracing.span import AgentExecutionSpan
 
@@ -1057,6 +1059,77 @@ async def _process_tool_call(
             should_yield = False
         return None, should_yield
 
+    @staticmethod
+    def _get_parallel_server_tool_requests(
+        config: Agent, state: AgentConversationExecutionState
+    ) -> List[ToolRequest]:
+        """Return queued requests that can safely be executed concurrently."""
+        if not config.llm.parallel_tool_calls or len(state.tool_call_queue) < 2:
+            return []
+
+        tools_by_name = {tool.name: tool for tool in (state.current_retrieved_tools or [])}
+        queued_requests = list(state.tool_call_queue)
+        queued_tools = [tools_by_name.get(tool_request.name) for tool_request in queued_requests]
+        if not all(
+            isinstance(tool, ServerTool) and not tool.requires_confirmation
+            for tool in queued_tools
+        ):
+            return []
+        return queued_requests
+
+    @staticmethod
+    async def _process_parallel_tool_calls(
+        agent_config: Agent,
+        agent_state: AgentConversationExecutionState,
+        conversation: "AgentConversation",
+        tool_requests: List[ToolRequest],
+    ) -> Optional[ExecutionStatus]:
+        """Execute an all-server-tool batch concurrently."""
+        # Remove the batch before starting it so a failure cannot cause a
+        # request to be executed twice if the conversation is resumed.
+        agent_state.tool_call_queue.clear()
+
+        results = await asyncio.gather(
+            *(
+                AgentConversationExecutor._execute_next_subcall(
+                    config=agent_config,
+                    conversation=conversation,
+                    state=agent_state,
+                    tool_request=tool_request,
+                    messages=conversation.message_list,
+                )
+                for tool_request in tool_requests
+            ),
+            return_exceptions=True,
+        )
+
+        raised_exceptions = [result for result in results if isinstance(result, Exception)]
+        if raised_exceptions:
+            raise raised_exceptions[0]
+
+        yielding_results = [
+            result
+            for result in results
+            if isinstance(result, ExecutionStatus) and result._requires_yielding
+        ]
+        if yielding_results:
+            # Server tools that are interrupted (for example by an AuthInterrupt) do not
+            # append a tool result message. Restore those requests so that resuming the
+            # conversation retries them. Requests which already have a result must stay
+            # removed from the queue, otherwise they would be executed twice.
+            completed_tool_request_ids = {
+                message.tool_result.tool_request_id
+                for message in conversation.message_list.messages
+                if message.tool_result is not None
+            }
+            agent_state.tool_call_queue.extend(
+                tool_request
+                for tool_request in tool_requests
+                if tool_request.tool_request_id not in completed_tool_request_ids
+            )
+            return yielding_results[0]
+        return None
+
     @staticmethod
     async def _execute_agent(
         conversation: "AgentConversation",
@@ -1095,7 +1168,24 @@ async def _execute_agent(
                 agent_state.current_tool_request = None
 
             elif len(agent_state.tool_call_queue) > 0:
-                agent_state._get_current_tool_request()
+                parallel_tool_requests = (
+                    AgentConversationExecutor._get_parallel_server_tool_requests(
+                        config=agent_config, state=agent_state
+                    )
+                )
+                if parallel_tool_requests:
+                    execution_status = (
+                        await AgentConversationExecutor._process_parallel_tool_calls(
+                            agent_config=agent_config,
+                            agent_state=agent_state,
+                            conversation=conversation,
+                            tool_requests=parallel_tool_requests,
+                        )
+                    )
+                    if execution_status is not None:
+                        return execution_status
+                else:
+                    agent_state._get_current_tool_request()
             elif agent_state.curr_iter >= agent_config.max_iterations:
                 if len(agent_config.output_descriptors) > 0:
                     default_outputs = _fill_submit_result_defaults(

wayflowcore/src/wayflowcore/models/geminimodel.py

@@ -129,6 +129,7 @@ def __init__(
         name: Optional[str] = None,
         description: Optional[str] = None,
         retry_policy: Optional[RetryPolicy] = None,
+        parallel_tool_calls: bool = True,
     ) -> None:
         self.auth = auth
         self.proxy = proxy
@@ -143,6 +144,7 @@ def __init__(
             generation_config=generation_config,
             supports_structured_generation=supports_structured_generation,
             supports_tool_calling=supports_tool_calling,
+            parallel_tool_calls=parallel_tool_calls,
             __metadata_info__=__metadata_info__,
             id=id,
             name=name,
@@ -289,6 +291,7 @@ def config(self) -> Dict[str, Any]:
             ),
             "supports_structured_generation": self.supports_structured_generation,
             "supports_tool_calling": self.supports_tool_calling,
+            **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}),
             "auth": self._serialize_auth_config(self.auth),
             "generation_config": (
                 self.generation_config.to_dict() if self.generation_config is not None else None

wayflowcore/src/wayflowcore/models/llmmodel.py

@@ -115,6 +115,7 @@ def __init__(
         id: Optional[str] = None,
         name: Optional[str] = None,
         description: Optional[str] = None,
+        parallel_tool_calls: bool = True,
     ):
         """
         Base class for LLM models.
@@ -137,6 +138,10 @@ def __init__(
             Whether the model supports tool calling or not. When set to `None`,
             the model will be prompted with a tool and it will check it can use
             the tool.
+        parallel_tool_calls:
+            Whether independent server-side tool calls returned in the same LLM
+            response should be executed concurrently. Client-side tools and
+            tools requiring confirmation are always handled sequentially.
         id:
             ID of the component.
         name:
@@ -178,6 +183,7 @@ def __init__(
             if supports_tool_calling is not None
             else _fetch_tool_calling_support(self)
         )
+        self.parallel_tool_calls = parallel_tool_calls
 
     @property
     def default_chat_template(self) -> "PromptTemplate":

wayflowcore/src/wayflowcore/models/ocigenaimodel.py

@@ -155,6 +155,7 @@ def __init__(
         auth_profile: Optional[str] = "DEFAULT",
         api_type: OciAPIType = OciAPIType.OCI,
         conversation_store_id: Optional[str] = None,
+        parallel_tool_calls: bool = True,
     ) -> None:
         """
         Model powered by OCIGenAI.
@@ -293,6 +294,7 @@ def __init__(
             generation_config=generation_config,
             supports_structured_generation=True,
             supports_tool_calling=True,
+            parallel_tool_calls=parallel_tool_calls,
             __metadata_info__=__metadata_info__,
         )
 
@@ -614,6 +616,7 @@ def config(self) -> Dict[str, Any]:
             "serving_mode": self.serving_mode.value,
             "compartment_id": self.compartment_id,
             "provider": self.provider.value,
+            **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}),
         }
 
     @property

wayflowcore/src/wayflowcore/models/ollamamodel.py

@@ -47,6 +47,7 @@ def __init__(
         name: Optional[str] = None,
         description: Optional[str] = None,
         retry_policy: Optional[RetryPolicy] = None,
+        parallel_tool_calls: bool = True,
     ) -> None:
         """
         Model powered by a locally hosted Ollama server.
@@ -119,6 +120,7 @@ def __init__(
             ca_file=ca_file,
             generation_config=generation_config,
             supports_tool_calling=supports_tool_calling,
+            parallel_tool_calls=parallel_tool_calls,
             supports_structured_generation=supports_structured_generation,
             __metadata_info__=__metadata_info__,
             id=id,
@@ -155,6 +157,7 @@ def config(self) -> Dict[str, Any]:
             "generation_config": (
                 self.generation_config.to_dict() if self.generation_config is not None else None
             ),
+            **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}),
         }
 
     @property

wayflowcore/src/wayflowcore/models/openaicompatiblemodel.py

@@ -59,6 +59,7 @@ def __init__(
         name: Optional[str] = None,
         description: Optional[str] = None,
         retry_policy: Optional[RetryPolicy] = None,
+        parallel_tool_calls: bool = True,
     ) -> None:
         """
         Model to use remote LLM endpoints that use OpenAI-compatible chat APIs.
@@ -135,6 +136,7 @@ def __init__(
             generation_config=generation_config,
             supports_structured_generation=supports_structured_generation,
             supports_tool_calling=supports_tool_calling,
+            parallel_tool_calls=parallel_tool_calls,
             __metadata_info__=__metadata_info__,
             id=id,
             name=name,
@@ -261,6 +263,7 @@ def config(self) -> Dict[str, Any]:
             ),
             "supports_structured_generation": self.supports_structured_generation,
             "supports_tool_calling": self.supports_tool_calling,
+            **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}),
             "generation_config": (
                 self.generation_config.to_dict() if self.generation_config is not None else None
             ),

wayflowcore/src/wayflowcore/models/openaimodel.py

@@ -29,6 +29,7 @@ def __init__(
         id: Optional[str] = None,
         name: Optional[str] = None,
         description: Optional[str] = None,
+        parallel_tool_calls: bool = True,
     ) -> None:
         """
         Model powered by OpenAI.
@@ -91,6 +92,7 @@ def __init__(
             proxy=proxy,
             api_key=api_key if api_key is not None else os.environ[OPEN_API_KEY],
             generation_config=generation_config,
+            parallel_tool_calls=parallel_tool_calls,
             supports_structured_generation=True,
             supports_tool_calling=True,
             api_type=api_type,
@@ -113,4 +115,5 @@ def config(self) -> Dict[str, Any]:
             "generation_config": (
                 self.generation_config.to_dict() if self.generation_config is not None else None
             ),
+            **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}),
         }

wayflowcore/src/wayflowcore/models/vllmmodel.py

@@ -44,6 +44,7 @@ def __init__(
         id: Optional[str] = None,
         name: Optional[str] = None,
         description: Optional[str] = None,
+        parallel_tool_calls: bool = True,
     ) -> None:
         """
         Model powered by a model hosted with VLLM server.
@@ -127,6 +128,7 @@ def __init__(
             generation_config=generation_config,
             supports_structured_generation=supports_structured_generation,
             supports_tool_calling=supports_tool_calling,
+            parallel_tool_calls=parallel_tool_calls,
             api_type=api_type,
             retry_policy=retry_policy,
             __metadata_info__=__metadata_info__,
@@ -148,6 +150,7 @@ def config(self) -> Dict[str, Any]:
             "generation_config": (
                 self.generation_config.to_dict() if self.generation_config is not None else None
             ),
+            **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}),
         }
 
     @property