feat: #1859 add runtime function tool concurrency config (#3152)

Author: seratch

src/agents/__init__.py

@@ -109,6 +109,7 @@
     Runner,
     ToolErrorFormatter,
     ToolErrorFormatterArgs,
+    ToolExecutionConfig,
 )
 from .run_context import AgentHookContext, RunContextWrapper, TContext
 from .run_error_handlers import (
@@ -432,6 +433,7 @@ def enable_verbose_stdout_logging():
     "ResponsesWebSocketSession",
     "RunConfig",
     "ReasoningItemIdPolicy",
+    "ToolExecutionConfig",
     "ToolErrorFormatter",
     "ToolErrorFormatterArgs",
     "RunState",

src/agents/run.py

@@ -40,6 +40,7 @@
     RunOptions,
     ToolErrorFormatter,
     ToolErrorFormatterArgs,
+    ToolExecutionConfig,
 )
 from .run_context import RunContextWrapper, TContext
 from .run_error_handlers import RunErrorHandlers
@@ -136,6 +137,7 @@
     "CallModelData",
     "CallModelInputFilter",
     "ReasoningItemIdPolicy",
+    "ToolExecutionConfig",
     "ToolErrorFormatter",
     "ToolErrorFormatterArgs",
     "DEFAULT_MAX_TURNS",

src/agents/run_config.py

@@ -88,6 +88,24 @@ class ToolErrorFormatterArgs(Generic[TContext]):
 ToolErrorFormatter = Callable[[ToolErrorFormatterArgs[Any]], MaybeAwaitable[str | None]]
 
 
+@dataclass
+class ToolExecutionConfig:
+    """Grouped SDK-side execution settings for local tool calls."""
+
+    max_function_tool_concurrency: int | None = None
+    """Maximum number of local function tool calls to execute concurrently.
+
+    Set to `None` to preserve the default behavior, which starts all function tool calls
+    emitted in a turn. This does not change provider-side `parallel_tool_calls` behavior.
+    """
+
+    def __post_init__(self) -> None:
+        if self.max_function_tool_concurrency is not None and (
+            self.max_function_tool_concurrency < 1
+        ):
+            raise ValueError("tool_execution.max_function_tool_concurrency must be at least 1")
+
+
 @dataclass
 class SandboxConcurrencyLimits:
     """Concurrency limits for sandbox materialization work."""
@@ -255,6 +273,9 @@ class RunConfig:
     sandbox: SandboxRunConfig | None = None
     """Optional sandbox runtime configuration for `SandboxAgent` execution."""
 
+    tool_execution: ToolExecutionConfig | None = None
+    """Optional SDK-side execution settings for local tool calls."""
+
 
 class RunOptions(TypedDict, Generic[TContext]):
     """Arguments for ``AgentRunner`` methods."""
@@ -297,6 +318,7 @@ class RunOptions(TypedDict, Generic[TContext]):
     "RunOptions",
     "SandboxConcurrencyLimits",
     "SandboxRunConfig",
+    "ToolExecutionConfig",
     "ToolErrorFormatter",
     "ToolErrorFormatterArgs",
     "_default_trace_include_sensitive_data",

src/agents/run_internal/tool_execution.py

@@ -1378,6 +1378,9 @@ def __init__(
         self.pending_tasks: set[asyncio.Task[Any]] = set()
         self.propagating_failure: BaseException | None = None
         self.available_function_tools: list[FunctionTool] = []
+        self.max_function_tool_concurrency = (
+            config.tool_execution.max_function_tool_concurrency if config.tool_execution else None
+        )
 
     async def execute(
         self,
@@ -1406,11 +1409,11 @@ async def execute(
             if function_tool_id not in enabled_function_tool_ids:
                 self.available_function_tools.append(tool_run.function_tool)
                 enabled_function_tool_ids.add(function_tool_id)
-        for order, tool_run in enumerate(self.tool_runs):
-            self._create_tool_task(tool_run, order)
+        pending_tool_runs = list(enumerate(self.tool_runs))
+        self._fill_tool_task_slots(pending_tool_runs)
 
         try:
-            await self._drain_pending_tasks()
+            await self._drain_pending_tasks(pending_tool_runs)
         except asyncio.CancelledError as exc:
             if self.propagating_failure is exc:
                 raise
@@ -1423,6 +1426,18 @@ async def execute(
             self.tool_output_guardrail_results,
         )
 
+    def _fill_tool_task_slots(self, pending_tool_runs: list[tuple[int, ToolRunFunction]]) -> None:
+        max_concurrency = self.max_function_tool_concurrency
+        available_slots = (
+            len(pending_tool_runs)
+            if max_concurrency is None
+            else max_concurrency - len(self.pending_tasks)
+        )
+        while available_slots > 0 and pending_tool_runs:
+            order, tool_run = pending_tool_runs.pop(0)
+            self._create_tool_task(tool_run, order)
+            available_slots -= 1
+
     def _create_tool_task(self, tool_run: ToolRunFunction, order: int) -> None:
         task_state = _FunctionToolTaskState(tool_run=tool_run, order=order)
         task = asyncio.create_task(
@@ -1435,7 +1450,10 @@ def _create_tool_task(self, tool_run: ToolRunFunction, order: int) -> None:
         self.task_states[task] = task_state
         self.pending_tasks.add(task)
 
-    async def _drain_pending_tasks(self) -> None:
+    async def _drain_pending_tasks(
+        self,
+        pending_tool_runs: list[tuple[int, ToolRunFunction]],
+    ) -> None:
         while self.pending_tasks:
             done_tasks, self.pending_tasks = await asyncio.wait(
                 self.pending_tasks,
@@ -1448,6 +1466,7 @@ async def _drain_pending_tasks(self) -> None:
             )
             if failure is not None:
                 await self._raise_failure_after_draining_siblings(failure)
+            self._fill_tool_task_slots(pending_tool_runs)
 
     async def _raise_failure_after_draining_siblings(
         self,

tests/test_run_config.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from agents import Agent, RunConfig, Runner
+from agents import Agent, RunConfig, Runner, ToolExecutionConfig
 from agents.model_settings import ModelSettings
 from agents.models.interface import Model, ModelProvider
 
@@ -185,3 +185,18 @@ def test_trace_include_sensitive_data_explicit_override_takes_precedence(monkeyp
     monkeypatch.setenv("OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA", "true")
     config = RunConfig(trace_include_sensitive_data=False)
     assert config.trace_include_sensitive_data is False
+
+
+def test_tool_execution_config_rejects_invalid_function_tool_concurrency() -> None:
+    with pytest.raises(
+        ValueError,
+        match="tool_execution.max_function_tool_concurrency must be at least 1",
+    ):
+        ToolExecutionConfig(max_function_tool_concurrency=0)
+
+
+def test_tool_execution_config_is_public_from_agents_package() -> None:
+    config = RunConfig(tool_execution=ToolExecutionConfig(max_function_tool_concurrency=2))
+
+    assert config.tool_execution is not None
+    assert config.tool_execution.max_function_tool_concurrency == 2

tests/test_run_step_execution.py

@@ -37,6 +37,7 @@
     ToolApprovalItem,
     ToolCallItem,
     ToolCallOutputItem,
+    ToolExecutionConfig,
     ToolGuardrailFunctionOutput,
     ToolInputGuardrail,
     ToolOutputGuardrailData,
@@ -232,6 +233,122 @@ async def test_plaintext_agent_with_tool_call_is_run_again():
     assert isinstance(result.next_step, NextStepRunAgain)
 
 
+@pytest.mark.asyncio
+async def test_function_tool_concurrency_default_starts_all_calls():
+    active_count = 0
+    max_seen_count = 0
+
+    async def tracked_tool(value: int) -> str:
+        nonlocal active_count, max_seen_count
+        active_count += 1
+        max_seen_count = max(max_seen_count, active_count)
+        try:
+            await asyncio.sleep(0.01)
+            return f"ok-{value}"
+        finally:
+            active_count -= 1
+
+    tool = function_tool(tracked_tool, name_override="tracked_tool")
+    agent = Agent(name="test", tools=[tool])
+    response = ModelResponse(
+        output=[
+            get_function_tool_call("tracked_tool", json.dumps({"value": 1}), call_id="call_1"),
+            get_function_tool_call("tracked_tool", json.dumps({"value": 2}), call_id="call_2"),
+            get_function_tool_call("tracked_tool", json.dumps({"value": 3}), call_id="call_3"),
+        ],
+        usage=Usage(),
+        response_id="resp",
+    )
+
+    result = await get_execute_result(agent, response)
+
+    assert active_count == 0
+    assert max_seen_count == 3
+    assert_item_is_function_tool_call_output(result.generated_items[3], "ok-1")
+    assert_item_is_function_tool_call_output(result.generated_items[4], "ok-2")
+    assert_item_is_function_tool_call_output(result.generated_items[5], "ok-3")
+
+
+@pytest.mark.asyncio
+async def test_function_tool_concurrency_cap_limits_calls_and_preserves_output_order():
+    active_count = 0
+    max_seen_count = 0
+
+    async def tracked_tool(value: int) -> str:
+        nonlocal active_count, max_seen_count
+        active_count += 1
+        max_seen_count = max(max_seen_count, active_count)
+        try:
+            await asyncio.sleep(0.03 if value == 1 else 0.001)
+            return f"ok-{value}"
+        finally:
+            active_count -= 1
+
+    tool = function_tool(tracked_tool, name_override="tracked_tool")
+    agent = Agent(name="test", tools=[tool])
+    response = ModelResponse(
+        output=[
+            get_function_tool_call("tracked_tool", json.dumps({"value": 1}), call_id="call_1"),
+            get_function_tool_call("tracked_tool", json.dumps({"value": 2}), call_id="call_2"),
+            get_function_tool_call("tracked_tool", json.dumps({"value": 3}), call_id="call_3"),
+        ],
+        usage=Usage(),
+        response_id="resp",
+    )
+
+    result = await get_execute_result(
+        agent,
+        response,
+        run_config=RunConfig(tool_execution=ToolExecutionConfig(max_function_tool_concurrency=2)),
+    )
+
+    assert active_count == 0
+    assert max_seen_count == 2
+    assert_item_is_function_tool_call_output(result.generated_items[3], "ok-1")
+    assert_item_is_function_tool_call_output(result.generated_items[4], "ok-2")
+    assert_item_is_function_tool_call_output(result.generated_items[5], "ok-3")
+
+
+@pytest.mark.asyncio
+async def test_function_tool_concurrency_cap_leaves_queued_calls_unstarted_after_failure():
+    started_tools: list[str] = []
+
+    async def failing_tool() -> str:
+        started_tools.append("failing_tool")
+        raise RuntimeError("boom")
+
+    async def queued_tool() -> str:
+        started_tools.append("queued_tool")
+        return "should-not-run"
+
+    failing = function_tool(
+        failing_tool,
+        name_override="failing_tool",
+        failure_error_function=None,
+    )
+    queued = function_tool(queued_tool, name_override="queued_tool")
+    agent = Agent(name="test", tools=[failing, queued])
+    response = ModelResponse(
+        output=[
+            get_function_tool_call("failing_tool", "{}", call_id="call_1"),
+            get_function_tool_call("queued_tool", "{}", call_id="call_2"),
+        ],
+        usage=Usage(),
+        response_id="resp",
+    )
+
+    with pytest.raises(UserError, match="Error running tool failing_tool: boom"):
+        await get_execute_result(
+            agent,
+            response,
+            run_config=RunConfig(
+                tool_execution=ToolExecutionConfig(max_function_tool_concurrency=1)
+            ),
+        )
+
+    assert started_tools == ["failing_tool"]
+
+
 @pytest.mark.asyncio
 async def test_plaintext_agent_hosted_shell_items_without_message_runs_again():
     shell_tool = ShellTool(environment={"type": "container_auto"})

tests/test_source_compat_constructors.py

@@ -17,6 +17,7 @@
     RunResult,
     RunResultStreaming,
     SessionSettings,
+    ToolExecutionConfig,
     ToolGuardrailFunctionOutput,
     ToolInputGuardrailData,
     ToolOutputGuardrailData,
@@ -92,6 +93,42 @@ def test_run_config_reasoning_item_id_policy_positional_binding() -> None:
 
     assert config.session_settings == session_settings
     assert config.reasoning_item_id_policy == "omit"
+    assert config.sandbox is None
+    assert config.tool_execution is None
+
+
+def test_run_config_tool_execution_append_preserves_sandbox_position() -> None:
+    session_settings = SessionSettings(limit=123)
+    tool_execution = ToolExecutionConfig(max_function_tool_concurrency=2)
+    config = RunConfig(
+        None,
+        MultiProvider(),
+        None,
+        None,
+        False,
+        None,
+        None,
+        None,
+        False,
+        None,
+        True,
+        "Agent workflow",
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        session_settings,
+        "omit",
+        None,
+        tool_execution,
+    )
+
+    assert config.session_settings == session_settings
+    assert config.reasoning_item_id_policy == "omit"
+    assert config.sandbox is None
+    assert config.tool_execution is tool_execution
 
 
 def test_model_settings_context_management_append_preserves_retry_position() -> None: