From e3efb1e57d81802025067853e30d276d7bfece81 Mon Sep 17 00:00:00 2001
From: pratikwayase <pratikwayase@gmail.com>
Date: Wed, 5 Aug 2026 15:49:56 +0530
Subject: [PATCH 1/6] feat(core): add tool concurrency groups and sequential
 execution order

---
 .../packages/core/agent_framework/_tools.py   | 199 +++++++++++++-----
 .../packages/core/agent_framework/_types.py   |   4 +
 python/packages/core/tests/core/test_tools.py | 104 +++++++++
 3 files changed, 260 insertions(+), 47 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 1e4089808a..20f81fe5d8 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -321,6 +321,7 @@ def __init__(
         func: Callable[..., Any] | None = None,
         input_model: type[BaseModel] | Mapping[str, Any] | None = None,
         result_parser: Callable[[Any], str | list[Content]] | _SkipParsingSentinel | None = None,
+        concurrency_group: str | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize the FunctionTool.
@@ -335,10 +336,11 @@ def __init__(
             max_invocations: The maximum number of times this function can be invoked
                 across the **lifetime of this tool instance**. If None (default),
                 there is no limit. Should be at least 1. If the tool is called multiple
-                times in one iteration, those will execute, after that it will stop working. For example,
-                if max_invocations is 3 and the tool is called 5 times in a single iteration,
-                these will complete, but any subsequent calls to the tool (in the same or future iterations)
-                will raise a ToolException.
+                times in one iteration, those will execute, after that it will stop
+                working. For example, if max_invocations is 3 and the tool is called 5
+                times in a single iteration, these will complete, but any subsequent
+                calls to the tool (in the same or future iterations) will raise a
+                ToolException.
 
                 .. note::
                     This counter lives on the tool instance and is never automatically
@@ -349,30 +351,37 @@ def __init__(
                     ``FunctionInvocationConfiguration["max_function_calls"]``
                     for per-request limits instead.
 
-            max_invocation_exceptions: The maximum number of exceptions allowed during invocations.
-                If None, there is no limit. Should be at least 1.
+            max_invocation_exceptions: The maximum number of exceptions allowed
+                during invocations. If None, there is no limit. Should be at least 1.
             additional_properties: Additional properties to set on the function.
-            func: The function to wrap. When ``None``, creates a declaration-only tool
-                that has no implementation. Declaration-only tools are useful when you want
-                the agent to reason about tool usage without executing them, or when the
-                actual implementation exists elsewhere (e.g., client-side rendering).
-            input_model: The Pydantic model that defines the input parameters for the function.
-                This can also be a JSON schema dictionary.
-                If not provided and ``func`` is not ``None``, it will be inferred from
-                the function signature. When ``func`` is ``None`` and ``input_model`` is
-                not provided, the tool will use an empty input model (no parameters) in
-                its JSON schema. For declaration-only tools that should declare
-                parameters, explicitly provide ``input_model`` (either a Pydantic
-                ``BaseModel`` or a JSON schema dictionary) so the model can reason about
-                the expected arguments.
-            result_parser: An optional callable with signature ``Callable[[Any], str]`` that
-                overrides the default result parsing behavior. When provided, this callable
-                is used to convert the raw function return value to a string instead of the
-                built-in :meth:`parse_result` logic. Pass the :data:`SKIP_PARSING` sentinel
-                instead of a callable to opt out of parsing entirely; in that case
-                :meth:`invoke` returns the wrapped function's raw return value. Depending
-                on your function, it may be easiest to just do the serialization directly
-                in the function body rather than providing a custom ``result_parser``.
+            func: The function to wrap. When ``None``, creates a declaration-only
+                tool that has no implementation. Declaration-only tools are useful
+                when you want the agent to reason about tool usage without executing
+                them, or when the actual implementation exists elsewhere (e.g.,
+                client-side rendering).
+            input_model: The Pydantic model that defines the input parameters for the
+                function. This can also be a JSON schema dictionary.
+                If not provided and ``func`` is not ``None``, it will be inferred
+                from the function signature. When ``func`` is ``None`` and
+                ``input_model`` is not provided, the tool will use an empty input
+                model (no parameters) in its JSON schema. For declaration-only tools
+                that should declare parameters, explicitly provide ``input_model``
+                (either a Pydantic ``BaseModel`` or a JSON schema dictionary) so the
+                model can reason about the expected arguments.
+            result_parser: An optional callable with signature ``Callable[[Any], str]``
+                that overrides the default result parsing behavior. When provided,
+                this callable is used to convert the raw function return value to a
+                string instead of the built-in :meth:`parse_result` logic. Pass the
+                :data:`SKIP_PARSING` sentinel instead of a callable to opt out of
+                parsing entirely; in that case :meth:`invoke` returns the wrapped
+                function's raw return value. Depending on your function, it may be
+                easiest to just do the serialization directly in the function body
+                rather than providing a custom ``result_parser``.
+            concurrency_group: If provided, tool calls with the same
+                concurrency_group will execute sequentially in the order they were
+                invoked by the model. Tools without a group, or with different
+                groups, will execute concurrently. Useful for stateful tools with
+                write->read dependencies to prevent race conditions.
             **kwargs: Additional keyword arguments.
         """
         # Core attributes (formerly from BaseTool)
@@ -417,6 +426,7 @@ def __init__(
         self._invocation_duration_histogram = _default_histogram()
         self.type: Literal["function_tool"] = "function_tool"
         self.result_parser = result_parser
+        self.concurrency_group = concurrency_group
 
     def _discover_injected_parameters(self) -> None:
         """Inspect the wrapped function for runtime injection parameters."""
@@ -905,9 +915,11 @@ def to_json_schema_spec(self) -> dict[str, Any]:
     @override
     def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True) -> dict[str, Any]:
         as_dict = super().to_dict(exclude=exclude, exclude_none=exclude_none)
+        if not exclude or "concurrency_group" not in exclude:
+            as_dict["concurrency_group"] = self.concurrency_group
         if (exclude and "input_model" in exclude) or not self.input_model:
             return as_dict
-        as_dict["input_model"] = self.parameters()  # Use cached parameters()
+        as_dict["input_model"] = self.parameters()
         return as_dict
 
 
@@ -1144,6 +1156,7 @@ def tool(
     max_invocations: int | None = None,
     max_invocation_exceptions: int | None = None,
     additional_properties: dict[str, Any] | None = None,
+    concurrency_group: str | None = None,
     result_parser: Callable[[Any], str | list[Content]] | _SkipParsingSentinel | None = None,
 ) -> FunctionTool: ...
 
@@ -1160,6 +1173,7 @@ def tool(
     max_invocations: int | None = None,
     max_invocation_exceptions: int | None = None,
     additional_properties: dict[str, Any] | None = None,
+    concurrency_group: str | None = None,
     result_parser: Callable[[Any], str | list[Content]] | _SkipParsingSentinel | None = None,
 ) -> Callable[[Callable[..., Any]], FunctionTool]: ...
 
@@ -1175,6 +1189,7 @@ def tool(
     max_invocations: int | None = None,
     max_invocation_exceptions: int | None = None,
     additional_properties: dict[str, Any] | None = None,
+    concurrency_group: str | None = None,
     result_parser: Callable[[Any], str | list[Content]] | _SkipParsingSentinel | None = None,
 ) -> FunctionTool | Callable[[Callable[..., Any]], FunctionTool]:
     """Decorate a function to turn it into a FunctionTool that can be passed to models and executed automatically.
@@ -1219,6 +1234,11 @@ def tool(
         max_invocation_exceptions: The maximum number of exceptions allowed during invocations.
             If None, there is no limit, should be at least 1.
         additional_properties: Additional properties to set on the function.
+        concurrency_group: If provided, tool calls with the same
+            concurrency_group will execute sequentially in the order they were
+            invoked by the model. Tools without a group, or with different
+            groups, will execute concurrently. Useful for stateful tools with
+            write->read dependencies to prevent race conditions.
         result_parser: An optional callable with signature ``Callable[[Any], str]`` that
             overrides the default result parsing. When provided, this callable converts the
             raw function return value to a string instead of using the built-in
@@ -1319,6 +1339,7 @@ def wrapper(f: Callable[..., Any]) -> FunctionTool:
                 func=f,
                 input_model=schema,
                 result_parser=result_parser,
+                concurrency_group=concurrency_group,
             )
 
         return wrapper(func)
@@ -1384,6 +1405,7 @@ class FunctionInvocationConfiguration(TypedDict, total=False):
     terminate_on_unknown_calls: bool
     additional_tools: Sequence[FunctionTool]
     include_detailed_errors: bool
+    tool_execution_order: Literal["parallel", "sequential"]
 
 
 def normalize_function_invocation_configuration(
@@ -1397,6 +1419,7 @@ def normalize_function_invocation_configuration(
         "terminate_on_unknown_calls": False,
         "additional_tools": [],
         "include_detailed_errors": False,
+        "tool_execution_order": "parallel",
     }
     if config:
         normalized.update(config)
@@ -1845,23 +1868,48 @@ async def _try_execute_function_call_groups(
     # Only a fully executable batch reaches this point; run calls concurrently but retain per-call result groups.
     # Create each task inside a copied context so the active agent span is
     # preserved for every parallel tool invocation.
-    execution_tasks = [
-        contextvars.copy_context().run(
-            asyncio.create_task,
-            _execute_single_function_call(
-                function_call,
-                custom_args=custom_args,
-                config=config,
-                tool_map=tool_map,
-                invocation_session=invocation_session,
-                middleware_pipeline=middleware_pipeline,
-                live_tools=live_tools,
-            ),
-        )
-        for function_call in function_calls
-    ]
+    execution_order = config.get("tool_execution_order", "parallel")
+
+    groups: dict[str, list[int]] = {}
+    for idx, function_call in enumerate(function_calls):
+        group_key: str | None = None
+        if execution_order == "parallel":
+            tool_name = _underlying_function_call(function_call).name
+            if tool_name is not None:
+                tool = tool_map.get(tool_name)
+                if tool is not None:
+                    group_key = getattr(tool, "concurrency_group", None)
+        if group_key is None:
+            group_key = "__sequential_all__" if execution_order == "sequential" else f"__ungrouped_{idx}"
+        if group_key not in groups:
+            groups[group_key] = []
+        groups[group_key].append(idx)
+
+    ordered_results: list[tuple[list[Content], bool] | None] = [None] * len(function_calls)
+
+    async def _execute_group(indices: list[int]) -> None:
+        for idx in indices:
+            call = function_calls[idx]
+            ctx = contextvars.copy_context()
+            task = ctx.run(
+                asyncio.create_task,
+                _execute_single_function_call(
+                    call,
+                    custom_args=custom_args,
+                    config=config,
+                    tool_map=tool_map,
+                    invocation_session=invocation_session,
+                    middleware_pipeline=middleware_pipeline,
+                    live_tools=live_tools,
+                ),
+            )
+            res = await task
+            ordered_results[idx] = res
+
+    execution_tasks = [asyncio.create_task(_execute_group(indices)) for indices in groups.values()]
+
     try:
-        execution_results = await asyncio.gather(*execution_tasks)
+        await asyncio.gather(*execution_tasks)
     except BaseException:
         # A loud escape from one call (e.g. MiddlewareFailure aborting the run
         # fail-closed) fails the whole batch: cancel in-flight siblings and wait for
@@ -1875,8 +1923,57 @@ async def _try_execute_function_call_groups(
         await asyncio.gather(*execution_tasks, return_exceptions=True)
         raise
 
-    should_terminate = any(terminate for _, terminate in execution_results)
-    return [result_contents for result_contents, _ in execution_results], should_terminate
+    if any(result is None for result in ordered_results):
+        raise RuntimeError("Internal error: missing tool execution result(s).")
+
+    completed_results = cast(list[tuple[list[Content], bool]], ordered_results)
+    should_terminate = any(terminate for _, terminate in completed_results)
+    return [result_contents for result_contents, _ in completed_results], should_terminate
+
+    groups: dict[str, list[int]] = {}
+    for idx, function_call in enumerate(function_calls):
+        group_key: str | None = None
+
+        if execution_order == "parallel":
+            tool = tool_map.get(function_call.name)
+            if tool is not None:
+                group_key = getattr(tool, "concurrency_group", None)
+
+        if group_key is None:
+            group_key = "__sequential_all__" if execution_order == "sequential" else f"__ungrouped_{idx}"
+
+        if group_key not in groups:
+            groups[group_key] = []
+        groups[group_key].append(idx)
+
+    ordered_results: list[tuple[list[Content], bool] | None] = [None] * len(function_calls)
+
+    async def _execute_group(indices: list[int]) -> None:
+        for idx in indices:
+            call = function_calls[idx]
+            ctx = contextvars.copy_context()
+            task = ctx.run(
+                asyncio.create_task,
+                _execute_single_function_call(
+                    call,
+                    custom_args=custom_args,
+                    config=config,
+                    tool_map=tool_map,
+                    invocation_session=invocation_session,
+                    middleware_pipeline=middleware_pipeline,
+                    live_tools=live_tools,
+                ),
+            )
+            res = await task
+            ordered_results[idx] = res
+
+    execution_tasks = [asyncio.create_task(_execute_group(indices)) for indices in groups.values()]
+
+    await asyncio.gather(*execution_tasks)
+
+    # Safer extraction to prevent TypeError if a result is unexpectedly None
+    should_terminate = any(result[1] for result in ordered_results if result is not None)
+    return [result[0] for result in ordered_results if result is not None], should_terminate
 
 
 @dataclass
@@ -3624,11 +3721,19 @@ def get_response(
         raw_session = request_kwargs.get("session")
         invocation_session = raw_session if isinstance(raw_session, _AgentSession) else None
 
+        # Give the loop private mutable options and one shared run-local tool list for progressive tool changes.
+        # Make options mutable so we can update conversation_id during function invocation loop
+        mutable_options: dict[str, Any] = dict(options) if options else {}
+
         # Bind one executor with the run's custom arguments, middleware, configuration, and session.
+        request_config = dict(self.function_invocation_configuration)
+        if tool_exec_order := mutable_options.get("tool_execution_order"):
+            request_config["tool_execution_order"] = tool_exec_order
+
         execute_function_calls = partial(
             _execute_function_calls,
             custom_args=additional_function_arguments,
-            config=self.function_invocation_configuration,
+            config=request_config,
             invocation_session=invocation_session,
             middleware_pipeline=function_middleware_pipeline,
         )
diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py
index 5199964d92..50449655a3 100644
--- a/python/packages/core/agent_framework/_types.py
+++ b/python/packages/core/agent_framework/_types.py
@@ -3708,6 +3708,10 @@ class _ChatOptionsBase(TypedDict, total=False):
     tool_choice: ToolMode | Literal["auto", "required", "none"]
     allow_multiple_tool_calls: bool
 
+    # Dictates whether multiple tool calls in a single message batch
+    # are executed concurrently (parallel) or one-by-one (sequential).
+    tool_execution_order: Literal["parallel", "sequential"]
+
     # Response configuration
     response_format: type[BaseModel] | Mapping[str, Any] | None
 
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index 33fad82ddc..580cf7fa00 100644
--- a/python/packages/core/tests/core/test_tools.py
+++ b/python/packages/core/tests/core/test_tools.py
@@ -20,6 +20,7 @@
     _auto_invoke_function,
     _parse_annotation,
     _parse_inputs,
+    _try_execute_function_call_groups,
     normalize_function_invocation_configuration,
 )
 from agent_framework.observability import OtelAttr
@@ -1576,3 +1577,106 @@ def test_skip_parsing_is_singleton() -> None:
 
 
 # endregion
+
+
+def test_tool_decorator_accepts_concurrency_group():
+    """Test that the @tool decorator accepts and stores the concurrency_group parameter."""
+
+    @tool(name="grouped_tool", concurrency_group="file_system")
+    def grouped_tool(x: int) -> int:
+        return x
+
+    assert isinstance(grouped_tool, FunctionTool)
+    assert grouped_tool.concurrency_group == "file_system"
+
+
+def test_function_invocation_configuration_accepts_execution_order():
+    """Test that execution_order is accepted and defaults to 'parallel'."""
+    config_seq = normalize_function_invocation_configuration({"tool_execution_order": "sequential"})
+    assert config_seq["tool_execution_order"] == "sequential"
+
+    config_default = normalize_function_invocation_configuration(None)
+    assert config_default["tool_execution_order"] == "parallel"
+
+
+async def test_try_execute_function_call_groups_concurrency_group():
+    """Tools in the same concurrency_group execute sequentially; ungrouped tools run concurrently."""
+    execution_order = []
+
+    @tool(concurrency_group="files")
+    async def write_file(name: str):
+        execution_order.append("write_start")
+        await asyncio.sleep(0.05)
+        execution_order.append("write_end")
+        return f"wrote {name}"
+
+    @tool(concurrency_group="files")
+    async def read_file(name: str):
+        execution_order.append("read_start")
+        await asyncio.sleep(0.01)
+        execution_order.append("read_end")
+        return f"read {name}"
+
+    @tool()
+    async def ungrouped_tool():
+        execution_order.append("ungrouped_start")
+        await asyncio.sleep(0.02)
+        execution_order.append("ungrouped_end")
+        return "ungrouped"
+
+    # Create function call contents simulating a batch from the LLM
+    call_write = Content.from_function_call(call_id="1", name="write_file", arguments='{"name": "test"}')
+    call_read = Content.from_function_call(call_id="2", name="read_file", arguments='{"name": "test"}')
+    call_ungrouped = Content.from_function_call(call_id="3", name="ungrouped_tool", arguments="{}")
+
+    config = normalize_function_invocation_configuration(None)
+
+    results, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[call_write, call_read, call_ungrouped],
+        tools=[write_file, read_file, ungrouped_tool],
+        config=config,
+    )
+
+    assert not should_terminate
+    assert len(results) == 3
+
+    assert execution_order.index("write_end") < execution_order.index("read_start")
+    assert execution_order.index("ungrouped_start") < execution_order.index("write_end")
+
+
+async def test_try_execute_function_call_groups_sequential_config():
+    """When execution_order is 'sequential', ALL tools run one-by-one regardless of groups."""
+    execution_order = []
+
+    @tool()
+    async def tool_a():
+        execution_order.append("a_start")
+        await asyncio.sleep(0.03)
+        execution_order.append("a_end")
+        return "a"
+
+    @tool()
+    async def tool_b():
+        execution_order.append("b_start")
+        await asyncio.sleep(0.01)
+        execution_order.append("b_end")
+        return "b"
+
+    call_a = Content.from_function_call(call_id="1", name="tool_a", arguments="{}")
+    call_b = Content.from_function_call(call_id="2", name="tool_b", arguments="{}")
+
+    config = normalize_function_invocation_configuration({"tool_execution_order": "sequential"})
+
+    results, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[call_a, call_b],
+        tools=[tool_a, tool_b],
+        config=config,
+    )
+
+    assert not should_terminate
+    assert execution_order == ["a_start", "a_end", "b_start", "b_end"]
+
+
+# endregion

From b8b1f034fbdc0eca40349c14a15e3ce7939171ac Mon Sep 17 00:00:00 2001
From: pratikwayase <pratikwayase@gmail.com>
Date: Wed, 5 Aug 2026 16:23:05 +0530
Subject: [PATCH 2/6] fix: address copilot review feedback for tool execution
 order

---
 python/packages/core/agent_framework/_tools.py | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 20f81fe5d8..87926a0db1 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -915,7 +915,9 @@ def to_json_schema_spec(self) -> dict[str, Any]:
     @override
     def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True) -> dict[str, Any]:
         as_dict = super().to_dict(exclude=exclude, exclude_none=exclude_none)
-        if not exclude or "concurrency_group" not in exclude:
+        if (not exclude or "concurrency_group" not in exclude) and (
+            not exclude_none or self.concurrency_group is not None
+        ):
             as_dict["concurrency_group"] = self.concurrency_group
         if (exclude and "input_model" in exclude) or not self.input_model:
             return as_dict
@@ -1971,9 +1973,11 @@ async def _execute_group(indices: list[int]) -> None:
 
     await asyncio.gather(*execution_tasks)
 
-    # Safer extraction to prevent TypeError if a result is unexpectedly None
-    should_terminate = any(result[1] for result in ordered_results if result is not None)
-    return [result[0] for result in ordered_results if result is not None], should_terminate
+    if any(result is None for result in ordered_results):
+        raise RuntimeError("Internal error: missing tool execution result(s).")
+    completed_results = cast(list[tuple[list[Content], bool]], ordered_results)
+    should_terminate = any(terminate for _, terminate in completed_results)
+    return [result_contents for result_contents, _ in completed_results], should_terminate
 
 
 @dataclass
@@ -3727,7 +3731,7 @@ def get_response(
 
         # Bind one executor with the run's custom arguments, middleware, configuration, and session.
         request_config = dict(self.function_invocation_configuration)
-        if tool_exec_order := mutable_options.get("tool_execution_order"):
+        if tool_exec_order := mutable_options.pop("tool_execution_order", None):
             request_config["tool_execution_order"] = tool_exec_order
 
         execute_function_calls = partial(

From 7783ec2d9c37b19f0e0dfde08017210269de8083 Mon Sep 17 00:00:00 2001
From: pratikwayase <pratikwayase@gmail.com>
Date: Sat, 8 Aug 2026 00:20:08 +0530
Subject: [PATCH 3/6] fix: resolve concurrency groups and CI fails

---
 .../packages/core/agent_framework/_tools.py   | 36 ++++++++++---------
 python/packages/core/tests/core/test_tools.py |  4 +--
 2 files changed, 22 insertions(+), 18 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 87926a0db1..3be3ece1df 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -1802,7 +1802,8 @@ async def _try_execute_function_call_groups(
     has_declaration_only_call = False
     # A user-input pause takes precedence over unknown-call termination in mixed batches.
     for function_call in actionable_calls:
-        function_name = function_call.name
+        function_name = _underlying_function_call(function_call).name
+
         logger.debug(
             "Checking function call: type=%s, name=%s, in approval_tools=%s",
             function_call.type,
@@ -1937,9 +1938,11 @@ async def _execute_group(indices: list[int]) -> None:
         group_key: str | None = None
 
         if execution_order == "parallel":
-            tool = tool_map.get(function_call.name)
-            if tool is not None:
-                group_key = getattr(tool, "concurrency_group", None)
+            tool_name = _underlying_function_call(function_call).name
+            if tool_name is not None:
+                tool = tool_map.get(tool_name)
+                if tool is not None:
+                    group_key = getattr(tool, "concurrency_group", None)
 
         if group_key is None:
             group_key = "__sequential_all__" if execution_order == "sequential" else f"__ungrouped_{idx}"
@@ -2012,6 +2015,11 @@ async def _execute_function_calls(
     invocation_session: AgentSession | None = None,
     middleware_pipeline: FunctionMiddlewarePipeline | None = None,
 ) -> _FunctionExecutionBatch:
+
+    run_config = cast("FunctionInvocationConfiguration", dict(config) if config else {})
+    if custom_args and "tool_execution_order" in custom_args:
+        run_config["tool_execution_order"] = custom_args["tool_execution_order"]
+
     tools = _extract_tools(options)
     if not tools:
         return _FunctionExecutionBatch(result_groups=[])
@@ -2021,7 +2029,7 @@ async def _execute_function_calls(
         tools=tools,
         invocation_session=invocation_session,
         middleware_pipeline=middleware_pipeline,
-        config=config,
+        config=run_config,
     )
     return _FunctionExecutionBatch(
         result_groups=result_groups,
@@ -3725,26 +3733,22 @@ def get_response(
         raw_session = request_kwargs.get("session")
         invocation_session = raw_session if isinstance(raw_session, _AgentSession) else None
 
-        # Give the loop private mutable options and one shared run-local tool list for progressive tool changes.
-        # Make options mutable so we can update conversation_id during function invocation loop
-        mutable_options: dict[str, Any] = dict(options) if options else {}
-
         # Bind one executor with the run's custom arguments, middleware, configuration, and session.
-        request_config = dict(self.function_invocation_configuration)
-        if tool_exec_order := mutable_options.pop("tool_execution_order", None):
-            request_config["tool_execution_order"] = tool_exec_order
+        options = dict(options) if options else {}
+
+        if tool_exec_order := options.pop("tool_execution_order", None):
+            additional_function_arguments["tool_execution_order"] = tool_exec_order
+
+        mutable_options: dict[str, Any] = dict(options)
 
         execute_function_calls = partial(
             _execute_function_calls,
             custom_args=additional_function_arguments,
-            config=request_config,
+            config=self.function_invocation_configuration,
             invocation_session=invocation_session,
             middleware_pipeline=function_middleware_pipeline,
         )
 
-        # Give the loop private mutable options and one shared run-local tool list for progressive tool changes.
-        # Make options mutable so we can update conversation_id during function invocation loop
-        mutable_options: dict[str, Any] = dict(options) if options else {}
         # Remove additional_function_arguments from options passed to underlying chat client
         # It's for tool invocation only and not recognized by chat service APIs
         mutable_options.pop("additional_function_arguments", None)
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index 580cf7fa00..d84d523d38 100644
--- a/python/packages/core/tests/core/test_tools.py
+++ b/python/packages/core/tests/core/test_tools.py
@@ -1601,7 +1601,7 @@ def test_function_invocation_configuration_accepts_execution_order():
 
 async def test_try_execute_function_call_groups_concurrency_group():
     """Tools in the same concurrency_group execute sequentially; ungrouped tools run concurrently."""
-    execution_order = []
+    execution_order: list[str] = []
 
     @tool(concurrency_group="files")
     async def write_file(name: str):
@@ -1647,7 +1647,7 @@ async def ungrouped_tool():
 
 async def test_try_execute_function_call_groups_sequential_config():
     """When execution_order is 'sequential', ALL tools run one-by-one regardless of groups."""
-    execution_order = []
+    execution_order: list[str] = []
 
     @tool()
     async def tool_a():

From fb4cb4f75eb1ddecca70f60d06877984216f8869 Mon Sep 17 00:00:00 2001
From: pratikwayase <pratikwayase@gmail.com>
Date: Sat, 15 Aug 2026 15:51:37 +0530
Subject: [PATCH 4/6] fix(core): keep tool_execution_order in config to prevent
 nested agent override

---
 python/packages/core/agent_framework/_tools.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 3be3ece1df..1e8a50f70b 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -3735,16 +3735,18 @@ def get_response(
 
         # Bind one executor with the run's custom arguments, middleware, configuration, and session.
         options = dict(options) if options else {}
+        run_config = cast ("FunctionInvocationConfiguration", dict(self.function_invocation_configuration) if self.function_invocation_configuration else {})
+
 
         if tool_exec_order := options.pop("tool_execution_order", None):
-            additional_function_arguments["tool_execution_order"] = tool_exec_order
+            run_config["tool_execution_order"] = tool_exec_order
 
         mutable_options: dict[str, Any] = dict(options)
 
         execute_function_calls = partial(
             _execute_function_calls,
             custom_args=additional_function_arguments,
-            config=self.function_invocation_configuration,
+            config=run_config,
             invocation_session=invocation_session,
             middleware_pipeline=function_middleware_pipeline,
         )

From e55413014eb449d47a2a2795062d100061f13656 Mon Sep 17 00:00:00 2001
From: pratikwayase <pratikwayase@gmail.com>
Date: Thu, 20 Aug 2026 00:16:00 +0530
Subject: [PATCH 5/6] fix(core): enforce tool_execution_order precedence over
 custom_args

---
 python/packages/core/agent_framework/_tools.py | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 1e8a50f70b..728fb5cb0e 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -3735,11 +3735,20 @@ def get_response(
 
         # Bind one executor with the run's custom arguments, middleware, configuration, and session.
         options = dict(options) if options else {}
-        run_config = cast ("FunctionInvocationConfiguration", dict(self.function_invocation_configuration) if self.function_invocation_configuration else {})
-
+        run_config = cast(
+            "FunctionInvocationConfiguration",
+            dict(self.function_invocation_configuration) if self.function_invocation_configuration else {},
+        )
 
+        if not isinstance(options, dict):  # pragma: no cover
+            options = {}
         if tool_exec_order := options.pop("tool_execution_order", None):
             run_config["tool_execution_order"] = tool_exec_order
+            if additional_function_arguments and "tool_execution_order" in additional_function_arguments:
+                logger.debug(
+                    "overriding tool_execution_order from function_invocation_kwargs with explicit run option: %s",
+                    tool_exec_order,
+                )
 
         mutable_options: dict[str, Any] = dict(options)
 

From f6b1b75ae306a17c272c8b3dd0fafccb4b6c0aac Mon Sep 17 00:00:00 2001
From: pratikwayase <pratikwayase@gmail.com>
Date: Fri, 4 Sep 2026 20:15:12 +0530
Subject: [PATCH 6/6] refactor: replace tool_execution_order with
 allow_concurrent_invocation

---
 .../packages/core/agent_framework/_tools.py   | 186 ++++--------------
 .../packages/core/agent_framework/_types.py   |   4 +-
 python/packages/core/tests/core/test_tools.py |  73 +------
 3 files changed, 45 insertions(+), 218 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 728fb5cb0e..d46da51a25 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -321,7 +321,6 @@ def __init__(
         func: Callable[..., Any] | None = None,
         input_model: type[BaseModel] | Mapping[str, Any] | None = None,
         result_parser: Callable[[Any], str | list[Content]] | _SkipParsingSentinel | None = None,
-        concurrency_group: str | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize the FunctionTool.
@@ -377,11 +376,6 @@ def __init__(
                 function's raw return value. Depending on your function, it may be
                 easiest to just do the serialization directly in the function body
                 rather than providing a custom ``result_parser``.
-            concurrency_group: If provided, tool calls with the same
-                concurrency_group will execute sequentially in the order they were
-                invoked by the model. Tools without a group, or with different
-                groups, will execute concurrently. Useful for stateful tools with
-                write->read dependencies to prevent race conditions.
             **kwargs: Additional keyword arguments.
         """
         # Core attributes (formerly from BaseTool)
@@ -426,7 +420,6 @@ def __init__(
         self._invocation_duration_histogram = _default_histogram()
         self.type: Literal["function_tool"] = "function_tool"
         self.result_parser = result_parser
-        self.concurrency_group = concurrency_group
 
     def _discover_injected_parameters(self) -> None:
         """Inspect the wrapped function for runtime injection parameters."""
@@ -915,10 +908,6 @@ def to_json_schema_spec(self) -> dict[str, Any]:
     @override
     def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True) -> dict[str, Any]:
         as_dict = super().to_dict(exclude=exclude, exclude_none=exclude_none)
-        if (not exclude or "concurrency_group" not in exclude) and (
-            not exclude_none or self.concurrency_group is not None
-        ):
-            as_dict["concurrency_group"] = self.concurrency_group
         if (exclude and "input_model" in exclude) or not self.input_model:
             return as_dict
         as_dict["input_model"] = self.parameters()
@@ -1158,7 +1147,6 @@ def tool(
     max_invocations: int | None = None,
     max_invocation_exceptions: int | None = None,
     additional_properties: dict[str, Any] | None = None,
-    concurrency_group: str | None = None,
     result_parser: Callable[[Any], str | list[Content]] | _SkipParsingSentinel | None = None,
 ) -> FunctionTool: ...
 
@@ -1175,7 +1163,6 @@ def tool(
     max_invocations: int | None = None,
     max_invocation_exceptions: int | None = None,
     additional_properties: dict[str, Any] | None = None,
-    concurrency_group: str | None = None,
     result_parser: Callable[[Any], str | list[Content]] | _SkipParsingSentinel | None = None,
 ) -> Callable[[Callable[..., Any]], FunctionTool]: ...
 
@@ -1191,7 +1178,6 @@ def tool(
     max_invocations: int | None = None,
     max_invocation_exceptions: int | None = None,
     additional_properties: dict[str, Any] | None = None,
-    concurrency_group: str | None = None,
     result_parser: Callable[[Any], str | list[Content]] | _SkipParsingSentinel | None = None,
 ) -> FunctionTool | Callable[[Callable[..., Any]], FunctionTool]:
     """Decorate a function to turn it into a FunctionTool that can be passed to models and executed automatically.
@@ -1236,11 +1222,6 @@ def tool(
         max_invocation_exceptions: The maximum number of exceptions allowed during invocations.
             If None, there is no limit, should be at least 1.
         additional_properties: Additional properties to set on the function.
-        concurrency_group: If provided, tool calls with the same
-            concurrency_group will execute sequentially in the order they were
-            invoked by the model. Tools without a group, or with different
-            groups, will execute concurrently. Useful for stateful tools with
-            write->read dependencies to prevent race conditions.
         result_parser: An optional callable with signature ``Callable[[Any], str]`` that
             overrides the default result parsing. When provided, this callable converts the
             raw function return value to a string instead of using the built-in
@@ -1341,7 +1322,6 @@ def wrapper(f: Callable[..., Any]) -> FunctionTool:
                 func=f,
                 input_model=schema,
                 result_parser=result_parser,
-                concurrency_group=concurrency_group,
             )
 
         return wrapper(func)
@@ -1407,7 +1387,7 @@ class FunctionInvocationConfiguration(TypedDict, total=False):
     terminate_on_unknown_calls: bool
     additional_tools: Sequence[FunctionTool]
     include_detailed_errors: bool
-    tool_execution_order: Literal["parallel", "sequential"]
+    allow_concurrent_invocation: bool
 
 
 def normalize_function_invocation_configuration(
@@ -1421,7 +1401,7 @@ def normalize_function_invocation_configuration(
         "terminate_on_unknown_calls": False,
         "additional_tools": [],
         "include_detailed_errors": False,
-        "tool_execution_order": "parallel",
+        "allow_concurrent_invocation": True,
     }
     if config:
         normalized.update(config)
@@ -1871,116 +1851,39 @@ async def _try_execute_function_call_groups(
     # Only a fully executable batch reaches this point; run calls concurrently but retain per-call result groups.
     # Create each task inside a copied context so the active agent span is
     # preserved for every parallel tool invocation.
-    execution_order = config.get("tool_execution_order", "parallel")
-
-    groups: dict[str, list[int]] = {}
-    for idx, function_call in enumerate(function_calls):
-        group_key: str | None = None
-        if execution_order == "parallel":
-            tool_name = _underlying_function_call(function_call).name
-            if tool_name is not None:
-                tool = tool_map.get(tool_name)
-                if tool is not None:
-                    group_key = getattr(tool, "concurrency_group", None)
-        if group_key is None:
-            group_key = "__sequential_all__" if execution_order == "sequential" else f"__ungrouped_{idx}"
-        if group_key not in groups:
-            groups[group_key] = []
-        groups[group_key].append(idx)
-
-    ordered_results: list[tuple[list[Content], bool] | None] = [None] * len(function_calls)
-
-    async def _execute_group(indices: list[int]) -> None:
-        for idx in indices:
-            call = function_calls[idx]
-            ctx = contextvars.copy_context()
-            task = ctx.run(
-                asyncio.create_task,
-                _execute_single_function_call(
-                    call,
-                    custom_args=custom_args,
-                    config=config,
-                    tool_map=tool_map,
-                    invocation_session=invocation_session,
-                    middleware_pipeline=middleware_pipeline,
-                    live_tools=live_tools,
-                ),
-            )
-            res = await task
-            ordered_results[idx] = res
-
-    execution_tasks = [asyncio.create_task(_execute_group(indices)) for indices in groups.values()]
-
-    try:
-        await asyncio.gather(*execution_tasks)
-    except BaseException:
-        # A loud escape from one call (e.g. MiddlewareFailure aborting the run
-        # fail-closed) fails the whole batch: cancel in-flight siblings and wait for
-        # them so no new tool work starts after the loop is abandoned. Cancellation
-        # is cooperative — a synchronous tool body already running in a worker thread
-        # (asyncio.to_thread) cannot be interrupted and may complete its side effects,
-        # but its result is discarded with the batch and never reaches the transcript,
-        # the model, or history.
-        for task in execution_tasks:
-            task.cancel()
-        await asyncio.gather(*execution_tasks, return_exceptions=True)
-        raise
-
-    if any(result is None for result in ordered_results):
-        raise RuntimeError("Internal error: missing tool execution result(s).")
-
-    completed_results = cast(list[tuple[list[Content], bool]], ordered_results)
-    should_terminate = any(terminate for _, terminate in completed_results)
-    return [result_contents for result_contents, _ in completed_results], should_terminate
-
-    groups: dict[str, list[int]] = {}
-    for idx, function_call in enumerate(function_calls):
-        group_key: str | None = None
-
-        if execution_order == "parallel":
-            tool_name = _underlying_function_call(function_call).name
-            if tool_name is not None:
-                tool = tool_map.get(tool_name)
-                if tool is not None:
-                    group_key = getattr(tool, "concurrency_group", None)
-
-        if group_key is None:
-            group_key = "__sequential_all__" if execution_order == "sequential" else f"__ungrouped_{idx}"
-
-        if group_key not in groups:
-            groups[group_key] = []
-        groups[group_key].append(idx)
-
-    ordered_results: list[tuple[list[Content], bool] | None] = [None] * len(function_calls)
-
-    async def _execute_group(indices: list[int]) -> None:
-        for idx in indices:
-            call = function_calls[idx]
-            ctx = contextvars.copy_context()
-            task = ctx.run(
-                asyncio.create_task,
-                _execute_single_function_call(
-                    call,
-                    custom_args=custom_args,
-                    config=config,
-                    tool_map=tool_map,
-                    invocation_session=invocation_session,
-                    middleware_pipeline=middleware_pipeline,
-                    live_tools=live_tools,
-                ),
-            )
-            res = await task
-            ordered_results[idx] = res
-
-    execution_tasks = [asyncio.create_task(_execute_group(indices)) for indices in groups.values()]
-
-    await asyncio.gather(*execution_tasks)
+    allow_concurrent = config.get("allow_concurrent_invocation", True)
+    execution_results: list[tuple[list[Content], bool]] = []
+
+    async def _execute_single(call: Content) -> tuple[list[Content], bool]:
+        ctx = contextvars.copy_context()
+        return await ctx.run(
+            _execute_single_function_call,
+            call,
+            custom_args=custom_args,
+            config=config,
+            tool_map=tool_map,
+            invocation_session=invocation_session,
+            middleware_pipeline=middleware_pipeline,
+            live_tools=live_tools,
+        )
 
-    if any(result is None for result in ordered_results):
-        raise RuntimeError("Internal error: missing tool execution result(s).")
-    completed_results = cast(list[tuple[list[Content], bool]], ordered_results)
-    should_terminate = any(terminate for _, terminate in completed_results)
-    return [result_contents for result_contents, _ in completed_results], should_terminate
+    if allow_concurrent:
+        tasks = [asyncio.create_task(_execute_single(call)) for call in function_calls]
+        try:
+            execution_results = await asyncio.gather(*tasks)
+        except BaseException:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            raise
+    else:
+        for call in function_calls:
+            res = await _execute_single(call)
+            execution_results.append(res)
+            if res[1]:
+                break
+    should_terminate = any(terminate for _, terminate in execution_results)
+    return [result_contents for result_contents, _ in execution_results], should_terminate
 
 
 @dataclass
@@ -2017,8 +1920,10 @@ async def _execute_function_calls(
 ) -> _FunctionExecutionBatch:
 
     run_config = cast("FunctionInvocationConfiguration", dict(config) if config else {})
-    if custom_args and "tool_execution_order" in custom_args:
-        run_config["tool_execution_order"] = custom_args["tool_execution_order"]
+    if custom_args and "allow_concurrent_invocation" in custom_args:
+        if "allow_concurrent_invocation" not in run_config:
+            run_config["allow_concurrent_invocation"] = custom_args["allow_concurrent_invocation"]
+        custom_args.pop("allow_concurrent_invocation")
 
     tools = _extract_tools(options)
     if not tools:
@@ -3734,23 +3639,14 @@ def get_response(
         invocation_session = raw_session if isinstance(raw_session, _AgentSession) else None
 
         # Bind one executor with the run's custom arguments, middleware, configuration, and session.
-        options = dict(options) if options else {}
+        mutable_options: dict[str, Any] = dict(options) if options else {}
         run_config = cast(
             "FunctionInvocationConfiguration",
             dict(self.function_invocation_configuration) if self.function_invocation_configuration else {},
         )
 
-        if not isinstance(options, dict):  # pragma: no cover
-            options = {}
-        if tool_exec_order := options.pop("tool_execution_order", None):
-            run_config["tool_execution_order"] = tool_exec_order
-            if additional_function_arguments and "tool_execution_order" in additional_function_arguments:
-                logger.debug(
-                    "overriding tool_execution_order from function_invocation_kwargs with explicit run option: %s",
-                    tool_exec_order,
-                )
-
-        mutable_options: dict[str, Any] = dict(options)
+        if allow_concurrent := mutable_options.pop("allow_concurrent_invocation", None):
+            run_config["allow_concurrent_invocation"] = allow_concurrent
 
         execute_function_calls = partial(
             _execute_function_calls,
diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py
index 50449655a3..a3df776e42 100644
--- a/python/packages/core/agent_framework/_types.py
+++ b/python/packages/core/agent_framework/_types.py
@@ -3709,8 +3709,8 @@ class _ChatOptionsBase(TypedDict, total=False):
     allow_multiple_tool_calls: bool
 
     # Dictates whether multiple tool calls in a single message batch
-    # are executed concurrently (parallel) or one-by-one (sequential).
-    tool_execution_order: Literal["parallel", "sequential"]
+    # are executed concurrently (True, default) or one-by-one (False).
+    allow_concurrent_invocation: bool
 
     # Response configuration
     response_format: type[BaseModel] | Mapping[str, Any] | None
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index d84d523d38..a16d47d11d 100644
--- a/python/packages/core/tests/core/test_tools.py
+++ b/python/packages/core/tests/core/test_tools.py
@@ -1579,74 +1579,8 @@ def test_skip_parsing_is_singleton() -> None:
 # endregion
 
 
-def test_tool_decorator_accepts_concurrency_group():
-    """Test that the @tool decorator accepts and stores the concurrency_group parameter."""
-
-    @tool(name="grouped_tool", concurrency_group="file_system")
-    def grouped_tool(x: int) -> int:
-        return x
-
-    assert isinstance(grouped_tool, FunctionTool)
-    assert grouped_tool.concurrency_group == "file_system"
-
-
-def test_function_invocation_configuration_accepts_execution_order():
-    """Test that execution_order is accepted and defaults to 'parallel'."""
-    config_seq = normalize_function_invocation_configuration({"tool_execution_order": "sequential"})
-    assert config_seq["tool_execution_order"] == "sequential"
-
-    config_default = normalize_function_invocation_configuration(None)
-    assert config_default["tool_execution_order"] == "parallel"
-
-
-async def test_try_execute_function_call_groups_concurrency_group():
-    """Tools in the same concurrency_group execute sequentially; ungrouped tools run concurrently."""
-    execution_order: list[str] = []
-
-    @tool(concurrency_group="files")
-    async def write_file(name: str):
-        execution_order.append("write_start")
-        await asyncio.sleep(0.05)
-        execution_order.append("write_end")
-        return f"wrote {name}"
-
-    @tool(concurrency_group="files")
-    async def read_file(name: str):
-        execution_order.append("read_start")
-        await asyncio.sleep(0.01)
-        execution_order.append("read_end")
-        return f"read {name}"
-
-    @tool()
-    async def ungrouped_tool():
-        execution_order.append("ungrouped_start")
-        await asyncio.sleep(0.02)
-        execution_order.append("ungrouped_end")
-        return "ungrouped"
-
-    # Create function call contents simulating a batch from the LLM
-    call_write = Content.from_function_call(call_id="1", name="write_file", arguments='{"name": "test"}')
-    call_read = Content.from_function_call(call_id="2", name="read_file", arguments='{"name": "test"}')
-    call_ungrouped = Content.from_function_call(call_id="3", name="ungrouped_tool", arguments="{}")
-
-    config = normalize_function_invocation_configuration(None)
-
-    results, should_terminate = await _try_execute_function_call_groups(
-        custom_args={},
-        function_calls=[call_write, call_read, call_ungrouped],
-        tools=[write_file, read_file, ungrouped_tool],
-        config=config,
-    )
-
-    assert not should_terminate
-    assert len(results) == 3
-
-    assert execution_order.index("write_end") < execution_order.index("read_start")
-    assert execution_order.index("ungrouped_start") < execution_order.index("write_end")
-
-
 async def test_try_execute_function_call_groups_sequential_config():
-    """When execution_order is 'sequential', ALL tools run one-by-one regardless of groups."""
+    """When allow_concurrent_invocation is False, ALL tools run one-by-one."""
     execution_order: list[str] = []
 
     @tool()
@@ -1665,16 +1599,13 @@ async def tool_b():
 
     call_a = Content.from_function_call(call_id="1", name="tool_a", arguments="{}")
     call_b = Content.from_function_call(call_id="2", name="tool_b", arguments="{}")
-
-    config = normalize_function_invocation_configuration({"tool_execution_order": "sequential"})
-
+    config = normalize_function_invocation_configuration({"allow_concurrent_invocation": False})
     results, should_terminate = await _try_execute_function_call_groups(
         custom_args={},
         function_calls=[call_a, call_b],
         tools=[tool_a, tool_b],
         config=config,
     )
-
     assert not should_terminate
     assert execution_order == ["a_start", "a_end", "b_start", "b_end"]