Python: feat(core): add tool concurrency groups and sequential execution order

[pratik wayase (PratikWayase)]

PR Description

Motivation & Context

Currently, the framework executes all tool calls requested in a single assistant message concurrently. While this is a great default for independent calls, models routinely emit dependent calls in one batch (e.g., "write the file, then read it"). Because the framework lacked a batch-wide execution control, dependent reads could race still-running writes, leading to "not found" errors and contradictory agent states.

This PR closes that gap by providing a declarative, framework-level boolean to control batch-wide execution order, preventing stateful tool race conditions without relying on fragile, tool-side asyncio.Lock workarounds.

Note on PR History: This PR supersedes and closes #7523 and #7881. It incorporates all maintainer feedback from those previous iterations, specifically narrowing the scope to batch-wide execution control (removing concurrency_group per #7914), centralizing the configuration, and resolving all architectural and CI blockers raised by reviewers.

Description & Review Guide

• What are the major changes?

  1. Batch-Wide Execution Control: Added allow_concurrent_invocation: bool to FunctionInvocationConfiguration (defaulting to True to preserve existing parallel behavior). When set to False, tools in the same batch execute one-by-one.
  2. Sequential Termination Handling: In _try_execute_function_call_groups, when allow_concurrent_invocation is False and a call requests termination, the loop now injects "Skipped: a prior tool call in this batch requested termination." results for all subsequent calls, ensuring provider continuation history remains fully resolved.
  3. Live Configuration Evaluation: Replaced the static config snapshot with a config_provider: Callable[[], FunctionInvocationConfiguration] in _execute_function_calls. This ensures dynamically registered tools (e.g., AG-UI server-managed placeholders) are respected during execution.
  4. Server-Tool Boundary Hardening: Added _is_server_managed_tool() filter to _get_tool_map() and a guard in _execute_single_function_call to guarantee hosted tools are never executed locally.
  5. Approval Replay Ordering: Tracked original_index in a parallel array for hidden already-approved requests. This ensures that resumed calls merge by their original batch position, restoring model order instead of reversing dependent operations.
  6. Pyright Strict Compliance: Fixed type inference issues in approval state serialization by explicitly typing serialized_requests and avoiding polluting the Content dictionaries with framework metadata.
  7. Tests & Formatting: Added test_try_execute_function_call_groups_sequential_config to verify execution order, and applied minor formatting fixes to _loop.py and _agent.py.

• What is the impact of these changes?
  This is fully backward compatible. The default behavior remains parallel execution (True). It provides tool authors a safe, declarative way to handle stateful dependencies, fixes a critical server-tool execution bug, and ensures strict type safety.

• What do you want reviewers to focus on?
  Please review the execution logic in _try_execute_function_call_groups (_tools.py). Specifically, verify that:

  • The config_provider in FunctionInvocationLayer.get_response cleanly evaluates live state without leaking options to the underlying chat client.
  • The _is_server_managed_tool filter cleanly prevents the get_time_zone local execution bug.
  • When allow_concurrent_invocation is False, the loop correctly breaks, stops dequeuing calls, and injects terminal "Skipped" results for the suffix.
  • The original_index tracking in _store_already_approved_approval_requests and _pop_already_approved_approval_responses correctly restores model order during sequential approval resumes.

Related Issue

Fixes #7386
Supersedes and closes #7523
Supersedes and closes #7881
Design for deferred per-tool controls: #7914

Contribution Checklist

• The code builds clean without any errors or warnings
• All unit tests pass, and I have added new tests where possible
• The PR follows the Contribution Guidelines
• This PR is linked to an issue and there is no other open PR for this issue (see Related Issue above).
• This is not a Breaking Change. Default behavior remains allow_concurrent_invocation = True.

[Copilot]

🟡 Changes recommended

Hosted-call isolation, context isolation, and approval replay ordering remain incorrect.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds configurable sequential tool execution while preserving concurrent execution by default.

Changes:

• Adds batch-wide invocation ordering and termination handling.
• Hardens hosted-tool and approval replay handling.
• Adds sequential execution coverage and formatting updates.

File summaries

File	Description
python/packages/core/agent_framework/_tools.py	Implements execution policy, hosted-tool filtering, and approval replay ordering.
python/packages/core/tests/core/test_tools.py	Tests sequential invocation order.
python/packages/core/agent_framework/_workflows/_agent.py	Formatting-only update.
python/packages/core/agent_framework/_harness/_loop.py	Formatting-only updates.

Review details

• Files reviewed: 4/4 changed files
• Comments generated: 4
• Review effort level: Balanced

---

💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.

[Eduard van Valkenburg (eavanvalkenburg)]

A few concerns to address before this is ready.

[Eduard van Valkenburg (eavanvalkenburg)]

The task boundary should stay in sequential mode: awaiting a newly created task immediately is still sequential, while the child task isolates ContextVar mutations from the orchestration task and subsequent tool calls. However, ctx.run(...) here does not provide that isolation: because _execute_single_function_call is async, this only creates its coroutine under ctx; the body later runs in the _execute_single task context. asyncio.create_task is what supplies the isolation. Could we replace this wrapper with a _create_execution_task helper that preserves the prior contextvars.copy_context().run(asyncio.create_task, _execute_single_function_call(...)) pattern and use it in both branches? Please also add a ContextVar regression test; the current ordering test would still pass with a direct await and would not protect this behavior.

[Eduard van Valkenburg (eavanvalkenburg)]

Sorting only the responses currently available cannot preserve original batch order when an earlier approval is still unanswered. For [approval-required write (0), safe read (1), approval-required notify (2)], answering only notify still puts it in responses_to_execute, so it executes before write; the updated hidden-group logic merely keeps read waiting. In sequential mode, later approved calls need to remain queued until every earlier decision in the batch prefix has been resolved.

[Eduard van Valkenburg (eavanvalkenburg)]

This says an ordinary tool failure stops dequeuing subsequent calls, but _auto_invoke_function converts most exceptions into a function-result error and returns terminate=False, so the sequential loop continues. Only a termination signal currently stops and synthesizes skipped results. Please either document the implemented termination-only behavior or make ordinary failures stop the suffix as described.

[Eduard van Valkenburg (eavanvalkenburg)]

config_provider() is evaluated twice here. Besides being unnecessary, that undermines the idea of taking one live configuration snapshot if the provider can change between calls. Please evaluate it once, then copy that result, for example config = config_provider(); run_config = cast(..., dict(config) if config else {}).