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/9] 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 1e4089808ad..20f81fe5d8e 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 5199964d92c..50449655a3b 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 33fad82ddc6..580cf7fa009 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/9] 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 20f81fe5d8e..87926a0db10 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/9] 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 87926a0db10..3be3ece1dfa 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 580cf7fa009..d84d523d38a 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/9] 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 3be3ece1dfa..1e8a50f70b3 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/9] 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 1e8a50f70b3..728fb5cb0e3 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/9] 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 728fb5cb0e3..d46da51a258 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 50449655a3b..a3df776e427 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 d84d523d38a..a16d47d11d7 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"]
 

From 774e41772b1f3a35a37c50658c16936d74491dfb Mon Sep 17 00:00:00 2001
From: pratikwayase <pratikwayase@gmail.com>
Date: Thu, 10 Sep 2026 01:20:13 +0530
Subject: [PATCH 7/9] fix(core): resolve server-tool execution and sequential
 approval ordering

---
 .../core/agent_framework/_harness/_loop.py    |   7 +-
 .../packages/core/agent_framework/_tools.py   | 131 ++++++++++++------
 .../packages/core/agent_framework/_types.py   |   4 -
 .../core/agent_framework/_workflows/_agent.py |   6 +-
 4 files changed, 97 insertions(+), 51 deletions(-)

diff --git a/python/packages/core/agent_framework/_harness/_loop.py b/python/packages/core/agent_framework/_harness/_loop.py
index 62200518eea..7b14765868e 100644
--- a/python/packages/core/agent_framework/_harness/_loop.py
+++ b/python/packages/core/agent_framework/_harness/_loop.py
@@ -34,7 +34,9 @@
 from pydantic import BaseModel, Field
 from typing_extensions import Self
 
-from .._agents import _LOOP_ITERATION_TOKEN_KEY  # pyright: ignore[reportPrivateUsage] -- shared loop-turn marker, see _agents.py
+from .._agents import (
+    _LOOP_ITERATION_TOKEN_KEY,  # pyright: ignore[reportPrivateUsage] -- shared loop-turn marker, see _agents.py
+)
 from .._feature_stage import ExperimentalFeature, experimental
 from .._middleware import AgentContext, AgentMiddleware, MiddlewareTermination
 from .._sessions import SessionContext
@@ -490,8 +492,7 @@ async def _fire_turn_scoped_after_providers(
         if response is None or run_after is None:
             return
         if not any(
-            getattr(provider, "after_run_once_per_turn", False)
-            for provider in getattr(agent, "context_providers", [])
+            getattr(provider, "after_run_once_per_turn", False) for provider in getattr(agent, "context_providers", [])
         ):
             return
         session_context = SessionContext(
diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index d46da51a258..7b99ece399e 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -1439,6 +1439,11 @@ def _function_execution_error_result(
     )
 
 
+def _is_server_managed_tool(tool: FunctionTool) -> bool:
+    """Check if a tool is server-managed and should not be executed locally."""
+    return bool(tool.additional_properties and tool.additional_properties.get("server_label"))
+
+
 async def _auto_invoke_function(
     function_call_content: Content,
     custom_args: dict[str, Any] | None = None,
@@ -1652,7 +1657,7 @@ def _get_tool_map(
     return {
         tool_item.name: tool_item
         for tool_item in _ensure_unique_tool_names(tools)
-        if isinstance(tool_item, FunctionTool)
+        if isinstance(tool_item, FunctionTool) and not _is_server_managed_tool(tool_item)
     }
 
 
@@ -1680,6 +1685,20 @@ async def _execute_single_function_call(
     from ._sessions import _suspend_run_persistence_gate  # pyright: ignore[reportPrivateUsage]
     from ._types import Content
 
+    source_function_call = _underlying_function_call(function_call)
+    tool_name = source_function_call.name
+
+    if tool_name not in tool_map:
+        exc = KeyError(f'Function "{tool_name}" not found.')
+        return [
+            Content.from_function_result(
+                call_id=source_function_call.call_id,  # type: ignore[arg-type]
+                result=f'Error: Requested function "{tool_name}" not found.',
+                exception=str(exc),
+                additional_properties=source_function_call.additional_properties,
+            )
+        ], False
+
     try:
         # A run-persistence gate defers only the gated run's own persistence; nested
         # agent runs persist inline at their own boundaries. Run-identity ownership
@@ -1697,11 +1716,10 @@ async def _execute_single_function_call(
                 config=config,
                 live_tools=live_tools,
             )
-        return [result], False
+            return [result], False
     except MiddlewareTermination as exc:
         if isinstance(exc.result, Content):
             return [exc.result], True
-        source_function_call = _underlying_function_call(function_call)
         return [
             Content.from_function_result(
                 call_id=source_function_call.call_id,  # type: ignore[arg-type]
@@ -1709,7 +1727,6 @@ async def _execute_single_function_call(
             )
         ], True
     except UserInputRequiredException as exc:
-        source_function_call = _underlying_function_call(function_call)
         call_id = source_function_call.call_id
         propagated_contents = [item for item in exc.contents if isinstance(item, Content)] if exc.contents else []
         for item in propagated_contents:
@@ -1800,12 +1817,10 @@ async def _try_execute_function_call_groups(
         if config.get("terminate_on_unknown_calls", False) and function_name not in tool_map:
             raise KeyError(f'Error: Requested function "{function_name}" not found.')
     if requires_approval:
-        # Surface only the approvals the host must decide; session-backed safe siblings wait for that resume.
-        # approval can only be needed for Function Call Content, not Approval Responses.
         logger.debug("Returning visible function_approval_request contents and storing already-approved requests")
         visible_requests: list[Content] = []
-        already_approved_requests: list[Content] = []
-        for function_call in function_calls:
+        already_approved_requests: list[tuple[int, Content]] = []
+        for idx, function_call in enumerate(function_calls):
             if function_call.type != "function_call":
                 continue
             approval_request = Content.from_function_approval_request(
@@ -1828,7 +1843,7 @@ async def _try_execute_function_call_groups(
             if invocation_session is None:
                 visible_requests.append(approval_request)
                 continue
-            already_approved_requests.append(approval_request)
+            already_approved_requests.append((idx, approval_request))
         _store_already_approved_approval_requests(
             invocation_session,
             visible_requests,
@@ -1877,11 +1892,19 @@ async def _execute_single(call: Content) -> tuple[list[Content], bool]:
             await asyncio.gather(*tasks, return_exceptions=True)
             raise
     else:
-        for call in function_calls:
+        for idx, call in enumerate(function_calls):
             res = await _execute_single(call)
             execution_results.append(res)
             if res[1]:
+                for skipped in function_calls[idx + 1 :]:
+                    skipped_call = _underlying_function_call(skipped)
+                    skipped_result = Content.from_function_result(
+                        call_id=skipped_call.call_id,  # type: ignore[arg-type]
+                        result="Skipped: a prior tool call in this batch requested termination.",
+                    )
+                    execution_results.append(([skipped_result], False))
                 break
+
     should_terminate = any(terminate for _, terminate in execution_results)
     return [result_contents for result_contents, _ in execution_results], should_terminate
 
@@ -1914,20 +1937,19 @@ async def _execute_function_calls(
     custom_args: dict[str, Any],
     function_calls: list[Content],
     options: dict[str, Any] | None,
-    config: FunctionInvocationConfiguration,
+    config_provider: Callable[[], FunctionInvocationConfiguration],
     invocation_session: AgentSession | None = None,
     middleware_pipeline: FunctionMiddlewarePipeline | None = None,
 ) -> _FunctionExecutionBatch:
-
-    run_config = cast("FunctionInvocationConfiguration", dict(config) if config else {})
-    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")
+    run_config = cast(
+        "FunctionInvocationConfiguration",
+        dict(config_provider()) if config_provider() else {},
+    )
 
     tools = _extract_tools(options)
     if not tools:
         return _FunctionExecutionBatch(result_groups=[])
+
     result_groups, should_terminate = await _try_execute_function_call_groups(
         custom_args=custom_args,
         function_calls=function_calls,
@@ -2263,7 +2285,7 @@ def _bind_approval_responses_to_pending_requests(
 def _store_already_approved_approval_requests(
     invocation_session: AgentSession | None,
     visible_approval_requests: Sequence[Content],
-    already_approved_requests: Sequence[Content],
+    already_approved_requests: Sequence[tuple[int, Content] | Content],
 ) -> None:
     """Store hidden already-approved requests keyed by the visible approvals that resume the batch."""
     if not already_approved_requests:
@@ -2271,15 +2293,33 @@ def _store_already_approved_approval_requests(
     state = _get_tool_approval_state(invocation_session)
     if state is None:
         return
-    visible_ids = [request.id for request in visible_approval_requests if request.id]
+    visible_ids: list[str] = [request.id for request in visible_approval_requests if request.id]
     if not visible_ids:
         return
 
     existing_groups = state.get(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY)
-    pending_groups = list(cast(list[Any], existing_groups)) if isinstance(existing_groups, list) else []
+
+    pending_groups: list[dict[str, Any]] = (
+        cast("list[dict[str, Any]]", existing_groups) if isinstance(existing_groups, list) else []
+    )
+
+    serialized_requests: list[dict[str, Any]] = []
+    request_indices: list[int] = []
+    for i, item in enumerate(already_approved_requests):
+        idx: int
+        request: Content
+        if isinstance(item, tuple):
+            idx, request = item
+        else:
+            idx, request = i, item  
+            
+        serialized_requests.append(request.to_dict())
+        request_indices.append(idx)
+
     pending_groups.append({
         "approval_request_ids": visible_ids,
-        "approval_requests": [request.to_dict() for request in already_approved_requests],
+        "approval_requests": serialized_requests,
+        "approval_request_indices": request_indices,
     })
     state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = pending_groups
 
@@ -2288,7 +2328,7 @@ def _pop_already_approved_approval_responses(
     invocation_session: AgentSession | None,
     approval_response_ids: set[str],
 ) -> list[Content]:
-    """Pop already-approved requests for the visible approval ids being answered."""
+    """Pop already-approved requests for the visible approval ids being answered, preserving original order."""
     if not approval_response_ids:
         return []
     state = _get_tool_approval_state(invocation_session)
@@ -2297,34 +2337,50 @@ def _pop_already_approved_approval_responses(
     raw_groups = state.get(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY, [])
     if not isinstance(raw_groups, list):
         return []
-    typed_groups = cast(list[Any], raw_groups)
+    typed_groups = cast("list[Mapping[str, Any]]", raw_groups)
 
-    responses: list[Content] = []
-    remaining_groups: list[Any] = []
+    tagged_responses: list[tuple[int, Content]] = []
+    remaining_groups: list[Mapping[str, Any]] = []
     for raw_group in typed_groups:
         if not isinstance(raw_group, Mapping):
             continue
-        group = cast(Mapping[str, Any], raw_group)
+        group = raw_group
         raw_ids = group.get("approval_request_ids")
-        group_ids: set[str] = {str(item) for item in cast(list[Any], raw_ids)} if isinstance(raw_ids, list) else set()
+        group_ids: set[str] = {str(item) for item in cast("list[Any]", raw_ids)} if isinstance(raw_ids, list) else set()
         if group_ids.isdisjoint(approval_response_ids):
-            remaining_groups.append(raw_group)
+            remaining_groups.append(group)
             continue
         raw_requests = group.get("approval_requests")
         if not isinstance(raw_requests, list):
             continue
-        for raw_request in cast(list[Any], raw_requests):
-            request = _content_from_state(raw_request)
+            
+        raw_indices = group.get("approval_request_indices")
+        indices = cast("list[int]", raw_indices) if isinstance(raw_indices, list) else []
+        
+        for i, raw_request in enumerate(cast(list[Any], raw_requests)):
+            if isinstance(raw_request, Mapping) and "original_index" in raw_request:
+                req_map = cast("Mapping[str, Any]", raw_request)
+                original_index: int = int(req_map["original_index"])
+                request_dict: dict[str, Any] = dict(req_map)
+                request_dict.pop("original_index", None)
+                request = _content_from_state(request_dict)
+            else:
+                original_index = indices[i] if i < len(indices) else 0
+                request = _content_from_state(raw_request)
+
             if request is None or request.type != "function_approval_request":
                 continue
-            responses.append(request.to_function_approval_response(approved=True))
+            tagged_responses.append((original_index, request.to_function_approval_response(approved=True)))
+
+    tagged_responses.sort(key=lambda item: item[0])
+    responses = [content for _, content in tagged_responses]
+
     if remaining_groups:
         state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = remaining_groups
     else:
         state.pop(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY, None)
     return responses
 
-
 def _collect_approval_responses(
     messages: list[Message],
 ) -> dict[str, Content]:
@@ -3640,18 +3696,15 @@ def get_response(
 
         # Bind one executor with the run's custom arguments, middleware, configuration, and session.
         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 allow_concurrent := mutable_options.pop("allow_concurrent_invocation", None):
-            run_config["allow_concurrent_invocation"] = allow_concurrent
+        def _get_run_config() -> FunctionInvocationConfiguration:
+            base = dict(self.function_invocation_configuration) if self.function_invocation_configuration else {}
+            return cast("FunctionInvocationConfiguration", base)
 
         execute_function_calls = partial(
             _execute_function_calls,
             custom_args=additional_function_arguments,
-            config=run_config,
+            config_provider=_get_run_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 a3df776e427..5199964d92c 100644
--- a/python/packages/core/agent_framework/_types.py
+++ b/python/packages/core/agent_framework/_types.py
@@ -3708,10 +3708,6 @@ 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 (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/agent_framework/_workflows/_agent.py b/python/packages/core/agent_framework/_workflows/_agent.py
index 09bd592e922..353a1a2efbc 100644
--- a/python/packages/core/agent_framework/_workflows/_agent.py
+++ b/python/packages/core/agent_framework/_workflows/_agent.py
@@ -713,11 +713,7 @@ def _process_request_info_event(
         Note:
             Text requests use the function-call envelope so callers can reply with a matching function result.
         """
-        if (
-            isinstance(event.data, Content)
-            and event.data.user_input_request
-            and event.data.type != "text"
-        ):
+        if isinstance(event.data, Content) and event.data.user_input_request and event.data.type != "text":
             # Preserve specialized requests that callers already understand how to present.
             return event.data
 

From d2c1d18d62e47d0f258066ba87bbae80aab40363 Mon Sep 17 00:00:00 2001
From: pratikwayase <pratikwayase@gmail.com>
Date: Thu, 10 Sep 2026 09:25:13 +0530
Subject: [PATCH 8/9] fix(core): addressed reviewer comments and fix CI fails
 pre-commit hooks

---
 .../packages/core/agent_framework/_tools.py   | 28 +++++++++++++------
 1 file changed, 20 insertions(+), 8 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 4b382a0b014..955a2355832 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -1545,6 +1545,7 @@ def _is_server_managed_tool(tool: FunctionTool) -> bool:
     """Check if a tool is server-managed and should not be executed locally."""
     return bool(tool.additional_properties and tool.additional_properties.get("server_label"))
 
+
 def _finalize_function_result(
     *,
     call_id: str,
@@ -1850,8 +1851,7 @@ async def _execute_single_function_call(
     tool_name = source_function_call.name
 
     if (
-        source_function_call.additional_properties
-        and source_function_call.additional_properties.get("server_label")
+        source_function_call.additional_properties and source_function_call.additional_properties.get("server_label")
     ) or tool_name not in tool_map:
         exc = KeyError(f'Function "{tool_name}" not found.')
         return [
@@ -2070,7 +2070,7 @@ async def _execute_single(call: Content) -> tuple[list[Content], bool]:
             except BaseException:
                 task.cancel()
                 raise
-                
+
             execution_results.append(res)
             if res[1]:
                 for skipped in function_calls[idx + 1 :]:
@@ -2554,8 +2554,8 @@ def _store_already_approved_approval_requests(
         if isinstance(item, tuple):
             idx, request = item
         else:
-            idx, request = i, item  
-            
+            idx, request = i, item
+
         serialized_requests.append(request.to_dict())
         request_indices.append(idx)
 
@@ -2590,16 +2590,27 @@ def _pop_already_approved_approval_responses(
         group = raw_group
         raw_ids = group.get("approval_request_ids")
         group_ids: set[str] = {str(item) for item in cast("list[Any]", raw_ids)} if isinstance(raw_ids, list) else set()
-        if group_ids.isdisjoint(approval_response_ids):
+
+        answered_ids = group_ids.intersection(approval_response_ids)
+        if not answered_ids:
             remaining_groups.append(group)
             continue
+
+        remaining_visible_ids = list(group_ids - answered_ids)
+
+        if remaining_visible_ids:
+            updated_group = dict(group)
+            updated_group["approval_request_ids"] = remaining_visible_ids
+            remaining_groups.append(updated_group)
+            continue
+
         raw_requests = group.get("approval_requests")
         if not isinstance(raw_requests, list):
             continue
-            
+
         raw_indices = group.get("approval_request_indices")
         indices = cast("list[int]", raw_indices) if isinstance(raw_indices, list) else []
-        
+
         for i, raw_request in enumerate(cast(list[Any], raw_requests)):
             if isinstance(raw_request, Mapping) and "original_index" in raw_request:
                 req_map = cast("Mapping[str, Any]", raw_request)
@@ -2624,6 +2635,7 @@ def _pop_already_approved_approval_responses(
         state.pop(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY, None)
     return responses
 
+
 def _collect_approval_responses(
     messages: list[Message],
 ) -> dict[str, Content]:

From 8b50fa745e5e484d8338ec98c05e4e8d9233dbcf Mon Sep 17 00:00:00 2001
From: pratikwayase <pratikwayase@gmail.com>
Date: Fri, 11 Sep 2026 12:16:04 +0530
Subject: [PATCH 9/9] fix(core): address reviewer feedback for task isolation
 and config evaluation

---
 .../packages/core/agent_framework/_tools.py   | 52 +++++++++++--------
 python/packages/core/tests/core/test_tools.py | 33 ++++++++++++
 2 files changed, 64 insertions(+), 21 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index d5e6368cba8..0a9496ad742 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -1463,9 +1463,11 @@ class FunctionInvocationConfiguration(TypedDict, total=False):
     - ``allow_concurrent_invocation``: Dictates whether multiple tool calls in a
       single message batch are executed concurrently (``True``, default) or
       one-by-one (``False``). When set to ``False``, tools run sequentially. If a
-      call requests termination or fails, the loop immediately stops dequeuing
-      subsequent calls and safely skips them, ensuring provider continuation
-      history remains resolved.
+      call requests termination (e.g., via middleware), the loop immediately stops
+      dequeuing subsequent calls and safely skips them, ensuring provider
+      continuation history remains resolved. Ordinary tool failures do not stop
+      the loop; they are converted to error results and the next tool in the
+      batch is executed.
 
     Note:
         ``max_iterations``, ``max_function_calls``, and ``max_duration_seconds``
@@ -1999,6 +2001,7 @@ async def _try_execute_function_call_groups(
         logger.debug("Returning visible function_approval_request contents and storing already-approved requests")
         visible_requests: list[Content] = []
         already_approved_requests: list[tuple[int, Content]] = []
+        batch_id = str(uuid4())
         for idx, function_call in enumerate(function_calls):
             if function_call.type != "function_call":
                 continue
@@ -2006,7 +2009,8 @@ async def _try_execute_function_call_groups(
                 id=function_call.id or function_call.call_id,  # type: ignore[arg-type]
                 function_call=function_call,
             )
-            approval_request.additional_properties = {"original_index": idx}
+            approval_request.additional_properties["original_index"] = idx
+            approval_request.additional_properties["batch_id"] = batch_id
             tool_name = function_call.name
             if tool_name is None:
                 visible_requests.append(approval_request)
@@ -2051,22 +2055,24 @@ async def _try_execute_function_call_groups(
     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]:
+    def _create_execution_task(call: Content) -> asyncio.Task[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,
-            host_payload_budget=host_payload_budget,
+        return 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,
+                host_payload_budget=host_payload_budget,
+            ),
         )
 
     if allow_concurrent:
-        tasks = [asyncio.create_task(_execute_single(call)) for call in function_calls]
+        tasks = [_create_execution_task(call) for call in function_calls]
         try:
             execution_results = await asyncio.gather(*tasks)
         except BaseException:
@@ -2076,7 +2082,7 @@ async def _execute_single(call: Content) -> tuple[list[Content], bool]:
             raise
     else:
         for idx, call in enumerate(function_calls):
-            task = asyncio.create_task(_execute_single(call))
+            task = _create_execution_task(call)
             try:
                 res = await task
             except BaseException:
@@ -2147,9 +2153,10 @@ async def _execute_function_calls(
     middleware_pipeline: FunctionMiddlewarePipeline | None = None,
     host_payload_budget: _FunctionResultPayloadBudget | None = None,
 ) -> _FunctionExecutionBatch:
+    config = config_provider()
     run_config = cast(
         "FunctionInvocationConfiguration",
-        dict(config_provider()) if config_provider() else {},
+        dict(config) if config else {},
     )
 
     tools = _extract_tools(options)
@@ -2505,9 +2512,12 @@ def _bind_approval_response_to_pending_request(
         additional_properties=rebound_properties,
         raw_representation=response.raw_representation,
     )
-    # Propagate the original batch index so sequential execution can restore model order
-    if request.additional_properties and "original_index" in request.additional_properties:
-        rebound.additional_properties["original_index"] = request.additional_properties["original_index"]
+    if request.additional_properties:
+        if "original_index" in request.additional_properties:
+            rebound.additional_properties["original_index"] = request.additional_properties["original_index"]
+        if "batch_id" in request.additional_properties:
+            rebound.additional_properties["batch_id"] = request.additional_properties["batch_id"]
+
     if consume:
         pending.pop(request_key, None)
         _save_pending_approval_requests(invocation_session, pending)
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index a16d47d11d7..b8191bb5379 100644
--- a/python/packages/core/tests/core/test_tools.py
+++ b/python/packages/core/tests/core/test_tools.py
@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 import asyncio
+import contextvars
 import threading
 from typing import Annotated, Any, Literal, get_args, get_origin
 from unittest.mock import Mock
@@ -1610,4 +1611,36 @@ async def tool_b():
     assert execution_order == ["a_start", "a_end", "b_start", "b_end"]
 
 
+async def test_sequential_execution_isolates_contextvars() -> None:
+    """ContextVar mutations in one tool should not leak to subsequent tools in sequential mode.
+
+    Regression test requested by reviewer: ensures that `asyncio.create_task` is used
+    in the sequential execution path to isolate ContextVar mutations between tool calls.
+    """
+    test_var: contextvars.ContextVar[str | None] = contextvars.ContextVar("test_var", default=None)
+
+    @tool()
+    async def tool_a() -> str:
+        test_var.set("polluted_by_a")
+        return "a"
+
+    @tool()
+    async def tool_b() -> str:
+        return test_var.get() or "None"
+
+    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({"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 results[1][0].result == "None"
+
+
 # endregion