""" Tool execution helpers for the run pipeline. This module hosts execution-time helpers, approval plumbing, and payload coercion. Action classes live in tool_actions.py. """ from __future__ import annotations import asyncio import dataclasses import functools import inspect import json from collections.abc import Awaitable, Callable, Mapping, Sequence from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast from openai.types.responses import ResponseFunctionToolCall from openai.types.responses.response_input_item_param import ( ComputerCallOutputAcknowledgedSafetyCheck, ) from openai.types.responses.response_input_param import McpApprovalResponse from openai.types.responses.response_output_item import McpApprovalRequest from .._tool_identity import ( FunctionToolLookupKey, NamedToolLookupKey, build_function_tool_lookup_map, get_function_tool_lookup_key, get_function_tool_lookup_key_for_call, get_function_tool_trace_name, get_tool_call_namespace, get_tool_call_trace_name, is_deferred_top_level_function_tool, normalize_tool_call_for_function_tool, should_allow_bare_name_approval_alias, tool_trace_name, ) from ..agent import Agent from ..agent_tool_state import ( consume_agent_tool_run_result, get_agent_tool_state_scope, peek_agent_tool_run_result, ) from ..editor import ApplyPatchOperation, ApplyPatchResult from ..exceptions import ( AgentsException, ModelBehaviorError, ToolInputGuardrailTripwireTriggered, ToolOutputGuardrailTripwireTriggered, UserError, ) from ..items import ( ItemHelpers, MCPApprovalResponseItem, RunItem, RunItemBase, ToolApprovalItem, ToolCallOutputItem, ) from ..logger import logger from ..model_settings import ModelSettings from ..run_config import RunConfig, ToolErrorFormatterArgs from ..run_context import RunContextWrapper from ..tool import ( ApplyPatchTool, ComputerTool, ComputerToolSafetyCheckData, FunctionTool, FunctionToolResult, ShellActionRequest, ShellCallData, ShellCallOutcome, ShellCommandOutput, Tool, ToolOrigin, get_function_tool_origin, invoke_function_tool, maybe_invoke_function_tool_failure_error_function, resolve_computer, ) from ..tool_context import ToolContext from ..tool_guardrails import ( ToolInputGuardrailData, ToolInputGuardrailResult, ToolOutputGuardrailData, ToolOutputGuardrailResult, ) from ..tracing import Span, SpanError, function_span, get_current_trace from ..util import _coro, _error_tracing from ..util._approvals import evaluate_needs_approval_setting from ..util._types import MaybeAwaitable from ._asyncio_progress import get_function_tool_task_progress_deadline from .agent_bindings import AgentBindings, bind_public_agent from .approvals import append_approval_error_output from .items import ( REJECTION_MESSAGE, extract_mcp_request_id, extract_mcp_request_id_from_run, function_rejection_item, ) from .run_steps import ToolRunFunction from .tool_use_tracker import AgentToolUseTracker if TYPE_CHECKING: from ..lifecycle import RunHooks from .run_steps import ( ToolRunApplyPatchCall, ToolRunComputerAction, ToolRunCustom, ToolRunFunction, ToolRunLocalShellCall, ToolRunShellCall, ) __all__ = [ "maybe_reset_tool_choice", "initialize_computer_tools", "extract_tool_call_id", "coerce_shell_call", "parse_apply_patch_custom_input", "parse_apply_patch_function_args", "extract_apply_patch_call_id", "coerce_apply_patch_operation", "coerce_apply_patch_operations", "normalize_apply_patch_result", "is_apply_patch_name", "normalize_shell_output", "serialize_shell_output", "resolve_exit_code", "render_shell_outputs", "truncate_shell_outputs", "normalize_max_output_length", "normalize_shell_output_entries", "format_shell_error", "get_trace_tool_error", "with_tool_function_span", "build_litellm_json_tool_call", "process_hosted_mcp_approvals", "collect_manual_mcp_approvals", "index_approval_items_by_call_id", "should_keep_hosted_mcp_item", "resolve_approval_status", "resolve_approval_interruption", "resolve_approval_rejection_message", "function_needs_approval", "resolve_enabled_function_tools", "execute_function_tool_calls", "execute_custom_tool_calls", "execute_local_shell_calls", "execute_shell_calls", "execute_apply_patch_calls", "execute_computer_actions", "execute_approved_tools", ] REDACTED_TOOL_ERROR_MESSAGE = "Tool execution failed. Error details are redacted." TToolSpanResult = TypeVar("TToolSpanResult") _FUNCTION_TOOL_CANCELLED_DRAIN_SECONDS = 0.25 _FUNCTION_TOOL_CANCELLED_IMMEDIATE_STEP_LIMIT = 64 _FUNCTION_TOOL_POST_INVOKE_WAIT_SECONDS = 0.1 _FunctionToolFailureSource = Literal["direct", "cancelled_teardown", "post_invoke"] _FunctionToolSettlementWaiter = Callable[ [set[asyncio.Task[Any]], asyncio.AbstractEventLoop, float], Awaitable[bool], ] _FunctionToolBackgroundExceptionMessage = Callable[[BaseException], str | None] @dataclasses.dataclass(frozen=True) class _FunctionToolFailure: """A function-tool failure with ordering metadata for arbitration.""" error: BaseException order: int source: _FunctionToolFailureSource = "direct" @dataclasses.dataclass class _FunctionToolTaskState: """Mutable execution state tracked for each function-tool task in a batch.""" tool_run: ToolRunFunction order: int invoke_task: asyncio.Task[Any] | None = None in_post_invoke_phase: bool = False def _background_cleanup_task_exception_message(exc: BaseException) -> str | None: """Return the loop-level message for late sibling-cleanup failures.""" if isinstance(exc, asyncio.CancelledError): return None if isinstance(exc, Exception): return ( "Background function tool task raised during cancellation cleanup after failure " "propagation." ) return "Background function tool task raised a fatal exception." def _background_post_invoke_task_exception_message(exc: BaseException) -> str | None: """Return the loop-level message for late post-invoke failures.""" del exc return "Background function tool post-invoke task raised after failure propagation." def _parent_cancelled_task_exception_message(exc: BaseException) -> str | None: """Return the loop-level message for detached tasks after parent cancellation.""" if isinstance(exc, Exception): return None return "Background function tool task raised a fatal exception." def _consume_function_tool_task_result( task: asyncio.Task[Any], *, message_for_exception: _FunctionToolBackgroundExceptionMessage, ) -> None: """Report background task failures according to the provided reporting policy.""" if task.cancelled(): return exc = task.exception() if exc is None: return message = message_for_exception(exc) if message is None: return task.get_loop().call_exception_handler( { "message": message, "exception": exc, "task": task, } ) def _get_function_tool_failure_priority(error: BaseException) -> int: """Return the precedence used to arbitrate concurrent function-tool failures.""" if isinstance(error, asyncio.CancelledError): return 0 if isinstance(error, Exception): return 1 return 2 def _select_function_tool_failure( current_failure: _FunctionToolFailure | None, new_failure: _FunctionToolFailure | None, ) -> _FunctionToolFailure | None: """Keep the highest-priority failure, breaking ties by tool call order.""" if current_failure is None: return new_failure if new_failure is None: return current_failure current_priority = _get_function_tool_failure_priority(current_failure.error) new_priority = _get_function_tool_failure_priority(new_failure.error) if new_priority > current_priority: return new_failure if new_priority == current_priority and new_failure.order < current_failure.order: return new_failure return current_failure def _merge_late_function_tool_failure( current_failure: _FunctionToolFailure | None, late_failure: _FunctionToolFailure | None, ) -> _FunctionToolFailure | None: """Merge a late failure into the triggering failure without masking the root cause.""" if current_failure is None: return late_failure if late_failure is None: return current_failure current_priority = _get_function_tool_failure_priority(current_failure.error) late_priority = _get_function_tool_failure_priority(late_failure.error) if late_priority > current_priority: return late_failure if late_priority < current_priority: return current_failure if late_failure.source == "post_invoke" and current_failure.source != "post_invoke": return late_failure return current_failure def _cancel_function_tool_tasks(tasks: set[asyncio.Task[Any]]) -> None: """Cancel sibling function-tool tasks.""" for task in tasks: task.cancel() def _attach_function_tool_task_result_callbacks( tasks: set[asyncio.Task[Any]], *, message_for_exception: _FunctionToolBackgroundExceptionMessage, ) -> None: """Attach a shared loop-level reporter to a set of background function-tool tasks.""" callback = functools.partial( _consume_function_tool_task_result, message_for_exception=message_for_exception, ) for task in tasks: task.add_done_callback(callback) def _record_completed_function_tool_tasks( *, completed_tasks: Sequence[asyncio.Task[Any]], task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState], results_by_tool_run: dict[int, Any], failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None, ignore_cancelled_tasks: set[asyncio.Task[Any]] | None = None, ) -> _FunctionToolFailure | None: """Store finished task results and return the preferred failure, if any.""" failure: _FunctionToolFailure | None = None ordered_done_tasks = sorted(completed_tasks, key=lambda task: task_states[task].order) ignored_tasks = ignore_cancelled_tasks or set() failure_sources = failure_sources_by_task or {} for task in ordered_done_tasks: task_state = task_states[task] tool_run = task_state.tool_run try: results_by_tool_run[id(tool_run)] = task.result() except BaseException as exc: if task in ignored_tasks and isinstance(exc, asyncio.CancelledError): continue failure = _select_function_tool_failure( failure, _FunctionToolFailure( error=exc, order=task_state.order, source=failure_sources.get(task, "direct"), ), ) return failure def _collect_settled_function_tool_tasks( *, remaining_tasks: set[asyncio.Task[Any]], task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState], results_by_tool_run: dict[int, Any], failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None, ignore_cancelled_tasks: set[asyncio.Task[Any]] | None = None, ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]: """Remove completed tasks from the pending set and record their outcomes.""" settled_tasks = {task for task in remaining_tasks if task.done()} if not settled_tasks: return None, remaining_tasks new_failure = _record_completed_function_tool_tasks( completed_tasks=list(settled_tasks), task_states=task_states, results_by_tool_run=results_by_tool_run, failure_sources_by_task=failure_sources_by_task, ignore_cancelled_tasks=ignore_cancelled_tasks, ) return new_failure, remaining_tasks - settled_tasks async def _wait_for_cancelled_function_tool_task_progress( remaining_tasks: set[asyncio.Task[Any]], loop: asyncio.AbstractEventLoop, remaining_time: float, *, task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState], ) -> tuple[bool, bool]: """Wait until a cancelled sibling can make another self-driven step.""" task_to_invoke_task = { tracked_task: task_state.invoke_task for tracked_task, task_state in task_states.items() if task_state.invoke_task is not None } progress_deadlines = { task: get_function_tool_task_progress_deadline( task=task, task_to_invoke_task=task_to_invoke_task, loop=loop, ) for task in remaining_tasks } self_progressing_tasks = { task: deadline for task, deadline in progress_deadlines.items() if deadline is not None } if not self_progressing_tasks: return False, False now = loop.time() next_deadline = min(self_progressing_tasks.values()) delay = max(0.0, next_deadline - now) if delay > 0: await asyncio.wait( set(self_progressing_tasks), timeout=min(delay, remaining_time), return_when=asyncio.FIRST_COMPLETED, ) return True, False await asyncio.sleep(0) return True, True async def _wait_for_function_tool_task_completion( remaining_tasks: set[asyncio.Task[Any]], _loop: asyncio.AbstractEventLoop, remaining_time: float, ) -> bool: """Wait briefly for a pending task to finish without forcing cancellation.""" done_tasks, _ = await asyncio.wait( remaining_tasks, timeout=remaining_time, return_when=asyncio.FIRST_COMPLETED, ) return bool(done_tasks) async def _settle_pending_function_tool_tasks( *, pending_tasks: set[asyncio.Task[Any]], task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState], results_by_tool_run: dict[int, Any], timeout_seconds: float, wait_for_pending_tasks: _FunctionToolSettlementWaiter, failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None, ignore_cancelled_tasks: set[asyncio.Task[Any]] | None = None, ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]: """Wait for pending tasks to settle within a bounded window and collect failures.""" if not pending_tasks: return None, set() failure: _FunctionToolFailure | None = None remaining_tasks = set(pending_tasks) loop = asyncio.get_running_loop() deadline = loop.time() + timeout_seconds while remaining_tasks: new_failure, remaining_tasks = _collect_settled_function_tool_tasks( remaining_tasks=remaining_tasks, task_states=task_states, results_by_tool_run=results_by_tool_run, failure_sources_by_task=failure_sources_by_task, ignore_cancelled_tasks=ignore_cancelled_tasks, ) failure = _select_function_tool_failure(failure, new_failure) if failure is not None and not isinstance(failure.error, Exception): break remaining_time = deadline - loop.time() if not remaining_tasks or remaining_time <= 0: break should_continue = await wait_for_pending_tasks(remaining_tasks, loop, remaining_time) if not should_continue: break new_failure, remaining_tasks = _collect_settled_function_tool_tasks( remaining_tasks=remaining_tasks, task_states=task_states, results_by_tool_run=results_by_tool_run, failure_sources_by_task=failure_sources_by_task, ignore_cancelled_tasks=ignore_cancelled_tasks, ) failure = _select_function_tool_failure(failure, new_failure) return failure, remaining_tasks async def _drain_cancelled_function_tool_tasks( *, pending_tasks: set[asyncio.Task[Any]], task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState], results_by_tool_run: dict[int, Any], failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None, ignore_cancelled_tasks: set[asyncio.Task[Any]] | None = None, ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]: """Drain cancelled siblings while they can continue making self-driven progress.""" remaining_immediate_steps = _FUNCTION_TOOL_CANCELLED_IMMEDIATE_STEP_LIMIT async def _wait_for_progress( remaining: set[asyncio.Task[Any]], loop: asyncio.AbstractEventLoop, remaining_time: float, ) -> bool: nonlocal remaining_immediate_steps if remaining_immediate_steps <= 0: return False ( should_continue, consumed_immediate_step, ) = await _wait_for_cancelled_function_tool_task_progress( remaining, loop, remaining_time, task_states=task_states, ) if consumed_immediate_step: remaining_immediate_steps -= 1 return should_continue return await _settle_pending_function_tool_tasks( pending_tasks=pending_tasks, task_states=task_states, results_by_tool_run=results_by_tool_run, timeout_seconds=_FUNCTION_TOOL_CANCELLED_DRAIN_SECONDS, wait_for_pending_tasks=_wait_for_progress, failure_sources_by_task=failure_sources_by_task, ignore_cancelled_tasks=ignore_cancelled_tasks, ) async def _wait_pending_function_tool_tasks_for_timeout( *, pending_tasks: set[asyncio.Task[Any]], task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState], results_by_tool_run: dict[int, Any], failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None, timeout_seconds: float, ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]: """Wait briefly for post-invoke siblings so in-flight failures can still surface.""" return await _settle_pending_function_tool_tasks( pending_tasks=pending_tasks, task_states=task_states, results_by_tool_run=results_by_tool_run, timeout_seconds=timeout_seconds, wait_for_pending_tasks=_wait_for_function_tool_task_completion, failure_sources_by_task=failure_sources_by_task, ) # -------------------------- # Public helpers # -------------------------- def maybe_reset_tool_choice( agent: Agent[Any], tool_use_tracker: AgentToolUseTracker, model_settings: ModelSettings, ) -> ModelSettings: """Reset tool_choice if the agent was forced to pick a tool previously and should be reset.""" if agent.reset_tool_choice is True and tool_use_tracker.has_used_tools(agent): return dataclasses.replace(model_settings, tool_choice=None) return model_settings async def resolve_enabled_function_tools( agent: Agent[Any], context_wrapper: RunContextWrapper[Any], ) -> list[FunctionTool]: """Resolve enabled function tools without triggering MCP tool discovery.""" async def _check_tool_enabled(tool: FunctionTool) -> bool: attr = tool.is_enabled if isinstance(attr, bool): return attr result = attr(context_wrapper, agent) if inspect.isawaitable(result): return bool(await result) return bool(result) function_tools = [tool for tool in agent.tools if isinstance(tool, FunctionTool)] if not function_tools: return [] enabled_results = await asyncio.gather(*(_check_tool_enabled(tool) for tool in function_tools)) return [tool for tool, enabled in zip(function_tools, enabled_results, strict=False) if enabled] async def initialize_computer_tools( *, tools: list[Tool], context_wrapper: RunContextWrapper[Any], ) -> None: """Resolve computer tools ahead of model invocation so each run gets its own instance.""" computer_tools = [tool for tool in tools if isinstance(tool, ComputerTool)] if not computer_tools: return await asyncio.gather( *(resolve_computer(tool=tool, run_context=context_wrapper) for tool in computer_tools) ) def get_mapping_or_attr(target: Any, key: str) -> Any: """Allow mapping-or-attribute access so tool payloads can be dicts or objects.""" if isinstance(target, Mapping): return target.get(key) return getattr(target, key, None) def extract_tool_call_id(raw: Any) -> str | None: """Return a call ID from tool call payloads or approval items.""" # OpenAI tool call payloads are documented to include a call_id/id so outputs can be matched. # See https://platform.openai.com/docs/guides/function-calling # We still guard against missing IDs to avoid hard failures on malformed or non-OpenAI inputs. if isinstance(raw, Mapping): candidate = raw.get("call_id") or raw.get("id") return candidate if isinstance(candidate, str) else None candidate = get_mapping_or_attr(raw, "call_id") or get_mapping_or_attr(raw, "id") return candidate if isinstance(candidate, str) else None def extract_shell_call_id(tool_call: Any) -> str: """Ensure shell calls include a call_id before executing them.""" value = extract_tool_call_id(tool_call) if not value: raise ModelBehaviorError("Shell call is missing call_id.") return str(value) def coerce_shell_call(tool_call: Any) -> ShellCallData: """Normalize a shell call payload into ShellCallData for consistent execution.""" call_id = extract_shell_call_id(tool_call) action_payload = get_mapping_or_attr(tool_call, "action") if action_payload is None: raise ModelBehaviorError("Shell call is missing an action payload.") commands_value = get_mapping_or_attr(action_payload, "commands") if not isinstance(commands_value, Sequence): raise ModelBehaviorError("Shell call action is missing commands.") commands: list[str] = [] for entry in commands_value: if entry is None: continue commands.append(str(entry)) if not commands: raise ModelBehaviorError("Shell call action must include at least one command.") timeout_value = ( get_mapping_or_attr(action_payload, "timeout_ms") or get_mapping_or_attr(action_payload, "timeoutMs") or get_mapping_or_attr(action_payload, "timeout") ) timeout_ms = int(timeout_value) if isinstance(timeout_value, int | float) else None max_length_value = get_mapping_or_attr(action_payload, "max_output_length") if max_length_value is None: max_length_value = get_mapping_or_attr(action_payload, "maxOutputLength") max_output_length = int(max_length_value) if isinstance(max_length_value, int | float) else None action = ShellActionRequest( commands=commands, timeout_ms=timeout_ms, max_output_length=max_output_length, ) status_value = get_mapping_or_attr(tool_call, "status") status_literal: Literal["in_progress", "completed"] | None = None if isinstance(status_value, str): lowered = status_value.lower() if lowered in {"in_progress", "completed"}: status_literal = cast(Literal["in_progress", "completed"], lowered) return ShellCallData(call_id=call_id, action=action, status=status_literal, raw=tool_call) def _parse_apply_patch_json(payload: str, *, label: str) -> dict[str, Any]: """Parse apply_patch JSON payloads with consistent error messages.""" try: parsed = json.loads(payload or "{}") except json.JSONDecodeError as exc: raise ModelBehaviorError(f"Invalid apply_patch {label} JSON: {exc}") from exc if not isinstance(parsed, Mapping): raise ModelBehaviorError(f"Apply patch {label} must be a JSON object.") return dict(parsed) def parse_apply_patch_custom_input(input_json: str) -> dict[str, Any]: """Parse custom apply_patch tool input used by legacy hosted-tool rollouts.""" parsed = _parse_apply_patch_json(input_json, label="input") if "operation" in parsed or "operations" in parsed: return parsed return {"operation": parsed} def parse_apply_patch_function_args(arguments: str) -> dict[str, Any]: """Parse apply_patch function tool arguments from the model.""" return _parse_apply_patch_json(arguments, label="arguments") def extract_apply_patch_call_id(tool_call: Any) -> str: """Ensure apply_patch calls include a call_id for approvals and tracing.""" value = extract_tool_call_id(tool_call) if not value: raise ModelBehaviorError("Apply patch call is missing call_id.") return str(value) def coerce_apply_patch_operation( tool_call: Any, *, context_wrapper: RunContextWrapper[Any] ) -> ApplyPatchOperation: """Normalize a single-operation tool payload for legacy callers.""" operations = coerce_apply_patch_operations(tool_call, context_wrapper=context_wrapper) if len(operations) != 1: raise ModelBehaviorError( f"Apply patch call includes {len(operations)} operations; expected exactly one." ) return operations[0] def coerce_apply_patch_operations( tool_call: Any, *, context_wrapper: RunContextWrapper[Any], ) -> list[ApplyPatchOperation]: """Normalize apply_patch payloads into one or more editor operations.""" raw_operations = get_mapping_or_attr(tool_call, "operations") if isinstance(raw_operations, list): operations = [ _coerce_apply_patch_operation_payload(operation, context_wrapper=context_wrapper) for operation in raw_operations ] if not operations: raise ModelBehaviorError("Apply patch call includes no operations.") return operations raw_operation = get_mapping_or_attr(tool_call, "operation") if raw_operation is not None: return [ _coerce_apply_patch_operation_payload(raw_operation, context_wrapper=context_wrapper) ] raise ModelBehaviorError("Apply patch call is missing an operation payload.") def _coerce_apply_patch_operation_payload( raw_operation: Any, *, context_wrapper: RunContextWrapper[Any] ) -> ApplyPatchOperation: """Normalize the tool payload into an ApplyPatchOperation the editor can consume.""" if raw_operation is None: raise ModelBehaviorError("Apply patch call is missing an operation payload.") op_type_value = str(get_mapping_or_attr(raw_operation, "type")) if op_type_value not in {"create_file", "update_file", "delete_file"}: raise ModelBehaviorError(f"Unknown apply_patch operation: {op_type_value}") op_type_literal = cast(Literal["create_file", "update_file", "delete_file"], op_type_value) path = get_mapping_or_attr(raw_operation, "path") if not isinstance(path, str) or not path: raise ModelBehaviorError("Apply patch operation is missing a valid path.") diff_value = get_mapping_or_attr(raw_operation, "diff") if op_type_literal in {"create_file", "update_file"}: if not isinstance(diff_value, str) or not diff_value: raise ModelBehaviorError( f"Apply patch operation {op_type_literal} is missing the required diff payload." ) diff: str | None = diff_value else: diff = None return ApplyPatchOperation( type=op_type_literal, path=str(path), diff=diff, ctx_wrapper=context_wrapper, move_to=_coerce_apply_patch_move_to(raw_operation), ) def _coerce_apply_patch_move_to(raw_operation: Any) -> str | None: move_to = get_mapping_or_attr(raw_operation, "move_to") if move_to is None: return None if not isinstance(move_to, str) or not move_to: raise ModelBehaviorError("Apply patch operation move_to must be a non-empty path.") return move_to def normalize_apply_patch_result( result: ApplyPatchResult | Mapping[str, Any] | str | None, ) -> ApplyPatchResult | None: """Coerce editor return values into ApplyPatchResult for consistent handling.""" if result is None: return None if isinstance(result, ApplyPatchResult): return result if isinstance(result, Mapping): status = result.get("status") output = result.get("output") normalized_status = status if status in {"completed", "failed"} else None normalized_output = str(output) if output is not None else None return ApplyPatchResult(status=normalized_status, output=normalized_output) if isinstance(result, str): return ApplyPatchResult(output=result) return ApplyPatchResult(output=str(result)) def is_apply_patch_name(name: str | None, tool: ApplyPatchTool | None) -> bool: """Allow flexible matching for apply_patch so existing names keep working.""" if not name: return False candidate = name.strip().lower() if candidate.startswith("apply_patch"): return True if tool and candidate == tool.name.strip().lower(): return True return False def normalize_shell_output(entry: ShellCommandOutput | Mapping[str, Any]) -> ShellCommandOutput: """Normalize shell output into ShellCommandOutput so downstream code sees a stable shape.""" if isinstance(entry, ShellCommandOutput): return entry stdout = str(entry.get("stdout", "") or "") stderr = str(entry.get("stderr", "") or "") command_value = entry.get("command") provider_data_value = entry.get("provider_data") outcome_value = entry.get("outcome") outcome_type: Literal["exit", "timeout"] = "exit" exit_code_value: Any | None = None if isinstance(outcome_value, Mapping): type_value = outcome_value.get("type") if type_value == "timeout": outcome_type = "timeout" elif isinstance(type_value, str): outcome_type = "exit" exit_code_value = outcome_value.get("exit_code") else: status_str = str(entry.get("status", "completed") or "completed").lower() if status_str == "timeout": outcome_type = "timeout" if isinstance(outcome_value, str): if outcome_value == "failure": exit_code_value = 1 elif outcome_value == "success": exit_code_value = 0 if exit_code_value is None and "exit_code" in entry: exit_code_value = entry.get("exit_code") outcome = ShellCallOutcome( type=outcome_type, exit_code=_normalize_exit_code(exit_code_value), ) return ShellCommandOutput( stdout=stdout, stderr=stderr, outcome=outcome, command=str(command_value) if command_value is not None else None, provider_data=cast(dict[str, Any], provider_data_value) if isinstance(provider_data_value, Mapping) else provider_data_value, ) def serialize_shell_output(output: ShellCommandOutput) -> dict[str, Any]: """Serialize ShellCommandOutput for persistence or cross-run transmission.""" payload: dict[str, Any] = { "stdout": output.stdout, "stderr": output.stderr, "status": output.status, "outcome": {"type": output.outcome.type}, } if output.outcome.type == "exit": payload["outcome"]["exit_code"] = output.outcome.exit_code if output.outcome.exit_code is not None: payload["exit_code"] = output.outcome.exit_code if output.command is not None: payload["command"] = output.command if output.provider_data: payload["provider_data"] = output.provider_data return payload def resolve_exit_code(raw_exit_code: Any, outcome_status: str | None) -> int: """Fallback logic to produce an exit code when providers omit one.""" normalized = _normalize_exit_code(raw_exit_code) if normalized is not None: return normalized normalized_status = (outcome_status or "").lower() if normalized_status == "success": return 0 if normalized_status == "failure": return 1 return 0 def render_shell_outputs(outputs: Sequence[ShellCommandOutput]) -> str: """Render shell outputs into human-readable text for tool responses.""" if not outputs: return "(no output)" rendered_chunks: list[str] = [] for result in outputs: chunk_lines: list[str] = [] if result.command: chunk_lines.append(f"$ {result.command}") stdout = result.stdout.rstrip("\n") stderr = result.stderr.rstrip("\n") if stdout: chunk_lines.append(stdout) if stderr: if stdout: chunk_lines.append("") chunk_lines.append("stderr:") chunk_lines.append(stderr) if result.exit_code not in (None, 0): chunk_lines.append(f"exit code: {result.exit_code}") if result.status == "timeout": chunk_lines.append("status: timeout") chunk = "\n".join(chunk_lines).strip() rendered_chunks.append(chunk if chunk else "(no output)") return "\n\n".join(rendered_chunks) def truncate_shell_outputs( outputs: Sequence[ShellCommandOutput], max_length: int ) -> list[ShellCommandOutput]: """Truncate shell output streams to a maximum combined length.""" if max_length <= 0: return [ ShellCommandOutput( stdout="", stderr="", outcome=output.outcome, command=output.command, provider_data=output.provider_data, ) for output in outputs ] remaining = max_length truncated: list[ShellCommandOutput] = [] for output in outputs: stdout = "" stderr = "" if remaining > 0 and output.stdout: stdout = output.stdout[:remaining] remaining -= len(stdout) if remaining > 0 and output.stderr: stderr = output.stderr[:remaining] remaining -= len(stderr) truncated.append( ShellCommandOutput( stdout=stdout, stderr=stderr, outcome=output.outcome, command=output.command, provider_data=output.provider_data, ) ) return truncated def normalize_shell_output_entries( entries: Sequence[Mapping[str, Any]], ) -> list[dict[str, Any]]: """Normalize raw shell output entries into the model-facing payload.""" structured_output: list[dict[str, Any]] = [] for entry in entries: sanitized = dict(entry) status_value = sanitized.pop("status", None) sanitized.pop("provider_data", None) raw_exit_code = sanitized.pop("exit_code", None) sanitized.pop("command", None) outcome_value = sanitized.get("outcome") if isinstance(outcome_value, str): resolved_type = "exit" if status_value == "timeout": resolved_type = "timeout" outcome_payload: dict[str, Any] = {"type": resolved_type} if resolved_type == "exit": outcome_payload["exit_code"] = resolve_exit_code(raw_exit_code, outcome_value) sanitized["outcome"] = outcome_payload elif isinstance(outcome_value, dict): outcome_payload = dict(outcome_value) outcome_status = outcome_payload.pop("status", None) outcome_type = outcome_payload.get("type") if outcome_type != "timeout": status_str = outcome_status if isinstance(outcome_status, str) else None outcome_payload.setdefault( "exit_code", resolve_exit_code(raw_exit_code, status_str), ) sanitized["outcome"] = outcome_payload structured_output.append(sanitized) return structured_output def normalize_max_output_length(value: int | None) -> int | None: """Clamp negative max output lengths to zero while preserving None.""" if value is None: return None return max(0, value) def format_shell_error(error: Exception | BaseException | Any) -> str: """Best-effort stringify of shell errors to keep tool failures readable.""" if isinstance(error, Exception): message = str(error) return message or error.__class__.__name__ try: return str(error) except Exception: # pragma: no cover - fallback only return repr(error) def get_trace_tool_error(*, trace_include_sensitive_data: bool, error_message: str) -> str: """Return a trace-safe tool error string based on the sensitive-data setting.""" return error_message if trace_include_sensitive_data else REDACTED_TOOL_ERROR_MESSAGE async def with_tool_function_span( *, config: RunConfig, tool_name: str, fn: Callable[[Span[Any] | None], MaybeAwaitable[TToolSpanResult]], ) -> TToolSpanResult: """Execute a tool callback in a function span when tracing is active.""" if config.tracing_disabled or get_current_trace() is None: result = fn(None) if inspect.isawaitable(result): return await result direct_result: object = result return cast(TToolSpanResult, direct_result) with function_span(tool_name) as span: result = fn(span) if inspect.isawaitable(result): return await result span_result: object = result return cast(TToolSpanResult, span_result) def build_litellm_json_tool_call(output: ResponseFunctionToolCall) -> FunctionTool: """Wrap a JSON string result in a FunctionTool so LiteLLM can stream it.""" async def on_invoke_tool(_ctx: ToolContext[Any], value: Any) -> Any: """Deserialize JSON strings so LiteLLM callers receive structured data.""" if isinstance(value, str): return json.loads(value) return value return FunctionTool( name=output.name, description=output.name, params_json_schema={}, on_invoke_tool=on_invoke_tool, strict_json_schema=True, is_enabled=True, _emit_tool_origin=False, ) async def resolve_approval_status( *, tool_name: str, call_id: str, raw_item: Any, agent: Agent[Any], context_wrapper: RunContextWrapper[Any], tool_namespace: str | None = None, tool_lookup_key: FunctionToolLookupKey | None = None, tool_origin: ToolOrigin | None = None, on_approval: Callable[[RunContextWrapper[Any], ToolApprovalItem], Any] | None = None, ) -> tuple[bool | None, ToolApprovalItem]: """Build approval item, run on_approval hook if needed, and return latest approval status.""" approval_item = ToolApprovalItem( agent=agent, raw_item=raw_item, tool_name=tool_name, tool_namespace=tool_namespace, tool_origin=tool_origin, tool_lookup_key=tool_lookup_key, ) approval_status = context_wrapper.get_approval_status( tool_name, call_id, tool_namespace=tool_namespace, existing_pending=approval_item, tool_lookup_key=tool_lookup_key, ) if approval_status is None and on_approval: decision_result = on_approval(context_wrapper, approval_item) if inspect.isawaitable(decision_result): decision_result = await decision_result if isinstance(decision_result, Mapping): if decision_result.get("approve") is True: context_wrapper.approve_tool(approval_item) elif decision_result.get("approve") is False: context_wrapper.reject_tool(approval_item) approval_status = context_wrapper.get_approval_status( tool_name, call_id, tool_namespace=tool_namespace, existing_pending=approval_item, tool_lookup_key=tool_lookup_key, ) return approval_status, approval_item def resolve_approval_interruption( approval_status: bool | None, approval_item: ToolApprovalItem, *, rejection_factory: Callable[[], RunItem], ) -> RunItem | ToolApprovalItem | None: """Return a rejection or pending approval item when approval is required.""" if approval_status is False: return rejection_factory() if approval_status is not True: return approval_item return None async def resolve_approval_rejection_message( *, context_wrapper: RunContextWrapper[Any], run_config: RunConfig, tool_type: Literal["function", "computer", "shell", "apply_patch", "custom"], tool_name: str, call_id: str, tool_namespace: str | None = None, tool_lookup_key: FunctionToolLookupKey | None = None, existing_pending: ToolApprovalItem | None = None, ) -> str: """Resolve model-visible output text for approval rejections.""" explicit_message = context_wrapper.get_rejection_message( tool_name, call_id, tool_namespace=tool_namespace, tool_lookup_key=tool_lookup_key, existing_pending=existing_pending, ) if explicit_message is not None: return explicit_message formatter = run_config.tool_error_formatter if formatter is None: return REJECTION_MESSAGE try: maybe_message = formatter( ToolErrorFormatterArgs( kind="approval_rejected", tool_type=tool_type, tool_name=tool_name, call_id=call_id, default_message=REJECTION_MESSAGE, run_context=context_wrapper, ) ) message = await maybe_message if inspect.isawaitable(maybe_message) else maybe_message except Exception as exc: logger.error("Tool error formatter failed for %s: %s", tool_name, exc) return REJECTION_MESSAGE if message is None: return REJECTION_MESSAGE if not isinstance(message, str): logger.error( "Tool error formatter returned non-string for %s: %s", tool_name, type(message).__name__, ) return REJECTION_MESSAGE return message async def function_needs_approval( function_tool: FunctionTool, context_wrapper: RunContextWrapper[Any], tool_call: ResponseFunctionToolCall, ) -> bool: """Evaluate a function tool's needs_approval setting with parsed args.""" parsed_args: dict[str, Any] = {} if callable(function_tool.needs_approval): try: parsed_args = json.loads(tool_call.arguments or "{}") except json.JSONDecodeError: parsed_args = {} needs_approval = await evaluate_needs_approval_setting( function_tool.needs_approval, context_wrapper, parsed_args, tool_call.call_id, ) return bool(needs_approval) def process_hosted_mcp_approvals( *, original_pre_step_items: Sequence[RunItem], mcp_approval_requests: Sequence[Any], context_wrapper: RunContextWrapper[Any], agent: Agent[Any], append_item: Callable[[RunItem], None], ) -> tuple[list[ToolApprovalItem], set[str]]: """Filter hosted MCP outputs and merge manual approvals so only coherent items remain.""" hosted_mcp_approvals_by_id: dict[str, ToolApprovalItem] = {} for item in original_pre_step_items: if not isinstance(item, ToolApprovalItem): continue raw = item.raw_item if not _is_hosted_mcp_approval_request(raw): continue request_id = extract_mcp_request_id(raw) if request_id: hosted_mcp_approvals_by_id[request_id] = item pending_hosted_mcp_approvals: list[ToolApprovalItem] = [] pending_hosted_mcp_approval_ids: set[str] = set() for mcp_run in mcp_approval_requests: request_id = extract_mcp_request_id_from_run(mcp_run) # MCP approval requests are documented to include an id used as approval_request_id. # See https://platform.openai.com/docs/guides/tools-connectors-mcp#approvals approval_item = hosted_mcp_approvals_by_id.get(request_id) if request_id else None if not approval_item or not request_id: continue tool_name = RunContextWrapper._resolve_tool_name(approval_item) approved = context_wrapper.get_approval_status( tool_name=tool_name, call_id=request_id, existing_pending=approval_item, ) if approved is not None: raw_item: McpApprovalResponse = { "type": "mcp_approval_response", "approval_request_id": request_id, "approve": approved, } rejection_message = context_wrapper.get_rejection_message( tool_name=tool_name, call_id=request_id, existing_pending=approval_item, ) if approved is False and rejection_message is not None: raw_item["reason"] = rejection_message response_item = MCPApprovalResponseItem(raw_item=raw_item, agent=agent) append_item(response_item) continue if approval_item not in pending_hosted_mcp_approvals: pending_hosted_mcp_approvals.append(approval_item) pending_hosted_mcp_approval_ids.add(request_id) append_item(approval_item) return pending_hosted_mcp_approvals, pending_hosted_mcp_approval_ids def collect_manual_mcp_approvals( *, agent: Agent[Any], requests: Sequence[Any], context_wrapper: RunContextWrapper[Any], existing_pending_by_call_id: Mapping[str, ToolApprovalItem] | None = None, ) -> tuple[list[MCPApprovalResponseItem], list[ToolApprovalItem]]: """Bridge hosted MCP approval requests with manual approvals to keep state consistent.""" pending_lookup = existing_pending_by_call_id or {} approved: list[MCPApprovalResponseItem] = [] pending: list[ToolApprovalItem] = [] seen_request_ids: set[str] = set() for request in requests: request_item = get_mapping_or_attr(request, "request_item") request_id = extract_mcp_request_id_from_run(request) # The Responses API returns mcp_approval_request items with an id to correlate approvals. # See https://platform.openai.com/docs/guides/tools-connectors-mcp#approvals if request_id and request_id in seen_request_ids: continue if request_id: seen_request_ids.add(request_id) tool_name = RunContextWrapper._to_str_or_none(getattr(request_item, "name", None)) tool_name = tool_name or get_mapping_or_attr(request, "mcp_tool").name existing_pending = pending_lookup.get(request_id or "") approval_status = context_wrapper.get_approval_status( tool_name, request_id or "", existing_pending=existing_pending ) if approval_status is not None and request_id: approval_response_raw: McpApprovalResponse = { "type": "mcp_approval_response", "approval_request_id": request_id, "approve": approval_status, } rejection_message = context_wrapper.get_rejection_message( tool_name, request_id, existing_pending=existing_pending, ) if approval_status is False and rejection_message is not None: approval_response_raw["reason"] = rejection_message approved.append(MCPApprovalResponseItem(raw_item=approval_response_raw, agent=agent)) continue if approval_status is not None: continue pending.append( existing_pending or ToolApprovalItem( agent=agent, raw_item=request_item, tool_name=tool_name, ) ) return approved, pending def index_approval_items_by_call_id(items: Sequence[RunItem]) -> dict[str, ToolApprovalItem]: """Build a mapping of tool call IDs to pending approval items.""" approvals: dict[str, ToolApprovalItem] = {} for item in items: if not isinstance(item, ToolApprovalItem): continue call_id = extract_tool_call_id(item.raw_item) if call_id: approvals[call_id] = item return approvals def should_keep_hosted_mcp_item( item: RunItem, *, pending_hosted_mcp_approvals: Sequence[ToolApprovalItem], pending_hosted_mcp_approval_ids: set[str], ) -> bool: """Keep only hosted MCP approvals that match pending requests from the provider.""" if not isinstance(item, ToolApprovalItem): return True if not _is_hosted_mcp_approval_request(item.raw_item): return False request_id = extract_mcp_request_id(item.raw_item) return item in pending_hosted_mcp_approvals or ( request_id is not None and request_id in pending_hosted_mcp_approval_ids ) class _FunctionToolBatchExecutor: """Own the mutable state needed to execute and arbitrate a function-tool batch.""" def __init__( self, *, bindings: AgentBindings[Any], tool_runs: list[ToolRunFunction], hooks: RunHooks[Any], context_wrapper: RunContextWrapper[Any], config: RunConfig, isolate_parallel_failures: bool | None, ) -> None: self.execution_agent = bindings.execution_agent self.public_agent = bindings.public_agent self.tool_runs = tool_runs self.hooks = hooks self.context_wrapper = context_wrapper self.config = config self.isolate_parallel_failures = ( len(tool_runs) > 1 if isolate_parallel_failures is None else isolate_parallel_failures ) self.tool_input_guardrail_results: list[ToolInputGuardrailResult] = [] self.tool_output_guardrail_results: list[ToolOutputGuardrailResult] = [] self.tool_state_scope_id = get_agent_tool_state_scope(context_wrapper) self.task_states: dict[asyncio.Task[Any], _FunctionToolTaskState] = {} self.teardown_cancelled_tasks: set[asyncio.Task[Any]] = set() self.results_by_tool_run: dict[int, Any] = {} self.pending_tasks: set[asyncio.Task[Any]] = set() self.propagating_failure: BaseException | None = None self.available_function_tools: list[FunctionTool] = [] async def execute( self, ) -> tuple[ list[FunctionToolResult], list[ToolInputGuardrailResult], list[ToolOutputGuardrailResult] ]: self.available_function_tools = await resolve_enabled_function_tools( self.execution_agent, self.context_wrapper, ) for tool_run in self.tool_runs: if tool_run.function_tool not in self.available_function_tools: self.available_function_tools.append(tool_run.function_tool) for order, tool_run in enumerate(self.tool_runs): self._create_tool_task(tool_run, order) try: await self._drain_pending_tasks() except asyncio.CancelledError as exc: if self.propagating_failure is exc: raise self._cancel_pending_tasks_for_parent_cancellation() raise return ( self._build_function_tool_results(), self.tool_input_guardrail_results, self.tool_output_guardrail_results, ) def _create_tool_task(self, tool_run: ToolRunFunction, order: int) -> None: task_state = _FunctionToolTaskState(tool_run=tool_run, order=order) task = asyncio.create_task( self._run_single_tool( task_state=task_state, func_tool=tool_run.function_tool, tool_call=tool_run.tool_call, ) ) self.task_states[task] = task_state self.pending_tasks.add(task) async def _drain_pending_tasks(self) -> None: while self.pending_tasks: done_tasks, self.pending_tasks = await asyncio.wait( self.pending_tasks, return_when=asyncio.FIRST_COMPLETED, ) failure = _record_completed_function_tool_tasks( completed_tasks=list(done_tasks), task_states=self.task_states, results_by_tool_run=self.results_by_tool_run, ) if failure is not None: await self._raise_failure_after_draining_siblings(failure) async def _raise_failure_after_draining_siblings( self, failure: _FunctionToolFailure, ) -> None: cancellable_tasks, post_invoke_tasks = self._partition_pending_tasks() self.teardown_cancelled_tasks.update(cancellable_tasks) _cancel_function_tool_tasks(cancellable_tasks) late_failure, remaining_cancelled_tasks = await self._drain_cancelled_tasks( cancellable_tasks ) post_invoke_failure, remaining_post_invoke_tasks = await self._wait_post_invoke_tasks( post_invoke_tasks ) _attach_function_tool_task_result_callbacks( remaining_cancelled_tasks, message_for_exception=_background_cleanup_task_exception_message, ) _attach_function_tool_task_result_callbacks( remaining_post_invoke_tasks, message_for_exception=_background_post_invoke_task_exception_message, ) merged_failure = _merge_late_function_tool_failure(failure, late_failure) merged_failure = _merge_late_function_tool_failure(merged_failure, post_invoke_failure) assert merged_failure is not None self.pending_tasks = set() self.propagating_failure = merged_failure.error raise merged_failure.error def _partition_pending_tasks(self) -> tuple[set[asyncio.Task[Any]], set[asyncio.Task[Any]]]: cancellable_tasks = { task for task in self.pending_tasks if not self.task_states[task].in_post_invoke_phase } return cancellable_tasks, self.pending_tasks - cancellable_tasks async def _drain_cancelled_tasks( self, tasks: set[asyncio.Task[Any]], ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]: late_failure_sources: dict[asyncio.Task[Any], _FunctionToolFailureSource] = dict.fromkeys( tasks, "cancelled_teardown", ) return await _drain_cancelled_function_tool_tasks( pending_tasks=tasks, task_states=self.task_states, results_by_tool_run=self.results_by_tool_run, failure_sources_by_task=late_failure_sources, ignore_cancelled_tasks=tasks, ) async def _wait_post_invoke_tasks( self, tasks: set[asyncio.Task[Any]], ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]: post_invoke_failure_sources: dict[asyncio.Task[Any], _FunctionToolFailureSource] = ( dict.fromkeys(tasks, "post_invoke") ) return await _wait_pending_function_tool_tasks_for_timeout( pending_tasks=tasks, task_states=self.task_states, results_by_tool_run=self.results_by_tool_run, failure_sources_by_task=post_invoke_failure_sources, timeout_seconds=_FUNCTION_TOOL_POST_INVOKE_WAIT_SECONDS, ) def _cancel_pending_tasks_for_parent_cancellation(self) -> None: self.teardown_cancelled_tasks.update(self.pending_tasks) _cancel_function_tool_tasks(self.pending_tasks) _attach_function_tool_task_result_callbacks( self.pending_tasks, message_for_exception=_parent_cancelled_task_exception_message, ) async def _run_single_tool( self, *, task_state: _FunctionToolTaskState, func_tool: FunctionTool, tool_call: ResponseFunctionToolCall, ) -> Any: raw_tool_call = tool_call outer_task = asyncio.current_task() task_state.in_post_invoke_phase = False tool_call = cast( ResponseFunctionToolCall, normalize_tool_call_for_function_tool(tool_call, func_tool), ) trace_tool_name = ( get_tool_call_trace_name(tool_call) or get_function_tool_trace_name(func_tool) or func_tool.name ) with function_span(trace_tool_name) as span_fn: tool_context_namespace = get_tool_call_namespace(raw_tool_call) if tool_context_namespace is None: tool_context_namespace = get_tool_call_namespace(tool_call) tool_context = ToolContext.from_agent_context( self.context_wrapper, tool_call.call_id, tool_call=raw_tool_call, tool_namespace=tool_context_namespace, agent=self.public_agent, run_config=self.config, ) agent_hooks = self.public_agent.hooks if self.config.trace_include_sensitive_data: span_fn.span_data.input = tool_call.arguments try: approval_result = await self._maybe_execute_tool_approval( func_tool=func_tool, tool_call=tool_call, raw_tool_call=raw_tool_call, span_fn=span_fn, ) if approval_result is not None: result = approval_result else: result = await self._execute_single_tool_body( outer_task=outer_task, task_state=task_state, func_tool=func_tool, tool_call=tool_call, tool_context=tool_context, agent_hooks=agent_hooks, ) except Exception as e: _error_tracing.attach_error_to_current_span( SpanError( message="Error running tool", data={"tool_name": func_tool.name, "error": str(e)}, ) ) if isinstance(e, AgentsException): raise e raise UserError(f"Error running tool {func_tool.name}: {e}") from e if self.config.trace_include_sensitive_data: span_fn.span_data.output = result return result async def _maybe_execute_tool_approval( self, *, func_tool: FunctionTool, tool_call: ResponseFunctionToolCall, raw_tool_call: ResponseFunctionToolCall, span_fn: Span[Any], ) -> Any | None: needs_approval_result = await function_needs_approval( func_tool, self.context_wrapper, tool_call, ) if not needs_approval_result: return None tool_namespace = get_tool_call_namespace(raw_tool_call) if tool_namespace is None and is_deferred_top_level_function_tool(func_tool): tool_namespace = func_tool.name tool_lookup_key = get_function_tool_lookup_key_for_call(raw_tool_call) if is_deferred_top_level_function_tool(func_tool): tool_lookup_key = ("deferred_top_level", func_tool.name) approval_status = self.context_wrapper.get_approval_status( func_tool.name, tool_call.call_id, tool_namespace=tool_namespace, tool_lookup_key=tool_lookup_key, ) if approval_status is None: approval_item = ToolApprovalItem( agent=self.public_agent, raw_item=raw_tool_call, tool_name=func_tool.name, tool_namespace=tool_namespace, tool_origin=get_function_tool_origin(func_tool), tool_lookup_key=tool_lookup_key, _allow_bare_name_alias=should_allow_bare_name_approval_alias( func_tool, self.available_function_tools, ), ) return FunctionToolResult(tool=func_tool, output=None, run_item=approval_item) if approval_status is not False: return None rejection_message = await resolve_approval_rejection_message( context_wrapper=self.context_wrapper, run_config=self.config, tool_type="function", tool_name=tool_trace_name(func_tool.name, tool_namespace) or func_tool.name, call_id=tool_call.call_id, tool_namespace=tool_namespace, tool_lookup_key=tool_lookup_key, ) span_fn.set_error( SpanError( message=rejection_message, data={ "tool_name": func_tool.name, "error": ( f"Tool execution for {tool_call.call_id} was manually rejected by user." ), }, ) ) span_fn.span_data.output = rejection_message return FunctionToolResult( tool=func_tool, output=rejection_message, run_item=function_rejection_item( self.public_agent, tool_call, rejection_message=rejection_message, scope_id=self.tool_state_scope_id, tool_origin=get_function_tool_origin(func_tool), ), ) async def _execute_single_tool_body( self, *, outer_task: asyncio.Task[Any] | None, task_state: _FunctionToolTaskState, func_tool: FunctionTool, tool_call: ResponseFunctionToolCall, tool_context: ToolContext[Any], agent_hooks: Any, ) -> Any: rejected_message = await _execute_tool_input_guardrails( func_tool=func_tool, tool_context=tool_context, agent=self.public_agent, tool_input_guardrail_results=self.tool_input_guardrail_results, ) if rejected_message is not None: return rejected_message await asyncio.gather( self.hooks.on_tool_start(tool_context, self.public_agent, func_tool), ( agent_hooks.on_tool_start(tool_context, self.public_agent, func_tool) if agent_hooks else _coro.noop_coroutine() ), ) invoke_task = asyncio.create_task( self._invoke_tool_and_run_post_invoke( outer_task=outer_task, task_state=task_state, func_tool=func_tool, tool_call=tool_call, tool_context=tool_context, agent_hooks=agent_hooks, ) ) task_state.invoke_task = invoke_task return await self._await_invoke_task(outer_task=outer_task, invoke_task=invoke_task) async def _invoke_tool_and_run_post_invoke( self, *, outer_task: asyncio.Task[Any] | None, task_state: _FunctionToolTaskState, func_tool: FunctionTool, tool_call: ResponseFunctionToolCall, tool_context: ToolContext[Any], agent_hooks: Any, ) -> Any: try: real_result = await invoke_function_tool( function_tool=func_tool, context=tool_context, arguments=tool_call.arguments, ) except asyncio.CancelledError as e: if outer_task in self.teardown_cancelled_tasks: raise result = await maybe_invoke_function_tool_failure_error_function( function_tool=func_tool, context=tool_context, error=e, ) if result is None: raise _error_tracing.attach_error_to_current_span( SpanError( message="Tool execution cancelled", data={"tool_name": func_tool.name, "error": str(e)}, ) ) real_result = result task_state.in_post_invoke_phase = True final_result = await _execute_tool_output_guardrails( func_tool=func_tool, tool_context=tool_context, agent=self.public_agent, real_result=real_result, tool_output_guardrail_results=self.tool_output_guardrail_results, ) await asyncio.gather( self.hooks.on_tool_end(tool_context, self.public_agent, func_tool, final_result), ( agent_hooks.on_tool_end(tool_context, self.public_agent, func_tool, final_result) if agent_hooks else _coro.noop_coroutine() ), ) return final_result async def _await_invoke_task( self, *, outer_task: asyncio.Task[Any] | None, invoke_task: asyncio.Task[Any], ) -> Any: try: return await asyncio.shield(invoke_task) except asyncio.CancelledError as cancel_exc: sibling_failure_cancelled = ( outer_task is not None and outer_task in self.teardown_cancelled_tasks ) if not invoke_task.done(): invoke_task.cancel() if sibling_failure_cancelled: invoke_results = await asyncio.gather(invoke_task, return_exceptions=True) invoke_failure = invoke_results[0] if invoke_results else None if isinstance(invoke_failure, BaseException) and not isinstance( invoke_failure, asyncio.CancelledError ): raise invoke_failure from cancel_exc elif invoke_task.done(): if not invoke_task.cancelled(): invoke_failure = invoke_task.exception() if isinstance(invoke_failure, BaseException) and not isinstance( invoke_failure, Exception ): raise invoke_failure from cancel_exc else: invoke_task.add_done_callback( functools.partial( _consume_function_tool_task_result, message_for_exception=_parent_cancelled_task_exception_message, ) ) raise def _get_nested_tool_interruptions( self, nested_run_result: Any | None, ) -> list[ToolApprovalItem]: """Extract nested approval interruptions from an agent tool run result.""" if nested_run_result is None or not hasattr(nested_run_result, "interruptions"): return [] return cast(list[ToolApprovalItem], nested_run_result.interruptions) def _consume_nested_tool_run_result( self, tool_run: ToolRunFunction, ) -> tuple[Any | None, list[ToolApprovalItem]]: """Consume stored nested run state for a tool call and return its interruptions.""" nested_run_result = consume_agent_tool_run_result( tool_run.tool_call, scope_id=self.tool_state_scope_id, ) return nested_run_result, self._get_nested_tool_interruptions(nested_run_result) def _resolve_nested_tool_run_result( self, tool_run: ToolRunFunction, ) -> tuple[Any | None, list[ToolApprovalItem]]: """Load nested run state, preserving unresolved interruptions until they are handled.""" nested_run_result = peek_agent_tool_run_result( tool_run.tool_call, scope_id=self.tool_state_scope_id, ) nested_interruptions = self._get_nested_tool_interruptions(nested_run_result) if nested_run_result is None or not nested_interruptions: nested_run_result, nested_interruptions = self._consume_nested_tool_run_result(tool_run) return nested_run_result, nested_interruptions def _build_function_tool_results(self) -> list[FunctionToolResult]: function_tool_results: list[FunctionToolResult] = [] for tool_run in self.tool_runs: result = self.results_by_tool_run[id(tool_run)] if isinstance(result, FunctionToolResult): nested_run_result, nested_interruptions = self._consume_nested_tool_run_result( tool_run ) if nested_run_result: result.agent_run_result = nested_run_result if nested_interruptions: result.interruptions = nested_interruptions function_tool_results.append(result) continue nested_run_result, nested_interruptions = self._resolve_nested_tool_run_result(tool_run) run_item: RunItem | None if not nested_interruptions: run_item = ToolCallOutputItem( output=result, raw_item=ItemHelpers.tool_call_output_item(tool_run.tool_call, result), agent=self.public_agent, tool_origin=get_function_tool_origin(tool_run.function_tool), ) else: # Skip tool output until nested interruptions are resolved. run_item = None function_tool_results.append( FunctionToolResult( tool=tool_run.function_tool, output=result, run_item=run_item, interruptions=nested_interruptions, agent_run_result=nested_run_result, ) ) return function_tool_results async def execute_function_tool_calls( *, bindings: AgentBindings[Any], tool_runs: list[ToolRunFunction], hooks: RunHooks[Any], context_wrapper: RunContextWrapper[Any], config: RunConfig, isolate_parallel_failures: bool | None = None, ) -> tuple[ list[FunctionToolResult], list[ToolInputGuardrailResult], list[ToolOutputGuardrailResult] ]: """Execute function tool calls with approvals, guardrails, and hooks.""" return await _FunctionToolBatchExecutor( bindings=bindings, tool_runs=tool_runs, hooks=hooks, context_wrapper=context_wrapper, config=config, isolate_parallel_failures=isolate_parallel_failures, ).execute() async def execute_custom_tool_calls( *, public_agent: Agent[Any], calls: list[ToolRunCustom], context_wrapper: RunContextWrapper[Any], hooks: RunHooks[Any], config: RunConfig, ) -> list[RunItem]: """Run Responses custom tool calls serially and wrap outputs.""" from .tool_actions import CustomToolAction results: list[RunItem] = [] for call in calls: results.append( await CustomToolAction.execute( agent=public_agent, call=call, hooks=hooks, context_wrapper=context_wrapper, config=config, ) ) return results async def execute_local_shell_calls( *, public_agent: Agent[Any], calls: list[ToolRunLocalShellCall], context_wrapper: RunContextWrapper[Any], hooks: RunHooks[Any], config: RunConfig, ) -> list[RunItem]: """Run local shell tool calls serially and wrap outputs.""" from .tool_actions import LocalShellAction results: list[RunItem] = [] for call in calls: results.append( await LocalShellAction.execute( agent=public_agent, call=call, hooks=hooks, context_wrapper=context_wrapper, config=config, ) ) return results async def execute_shell_calls( *, public_agent: Agent[Any], calls: list[ToolRunShellCall], context_wrapper: RunContextWrapper[Any], hooks: RunHooks[Any], config: RunConfig, ) -> list[RunItem]: """Run shell tool calls serially and wrap outputs.""" from .tool_actions import ShellAction results: list[RunItem] = [] for call in calls: results.append( await ShellAction.execute( agent=public_agent, call=call, hooks=hooks, context_wrapper=context_wrapper, config=config, ) ) return results async def execute_apply_patch_calls( *, public_agent: Agent[Any], calls: list[ToolRunApplyPatchCall], context_wrapper: RunContextWrapper[Any], hooks: RunHooks[Any], config: RunConfig, ) -> list[RunItem]: """Run apply_patch tool calls serially and normalize outputs.""" from .tool_actions import ApplyPatchAction results: list[RunItem] = [] for call in calls: results.append( await ApplyPatchAction.execute( agent=public_agent, call=call, hooks=hooks, context_wrapper=context_wrapper, config=config, ) ) return results async def execute_computer_actions( *, public_agent: Agent[Any], actions: list[ToolRunComputerAction], hooks: RunHooks[Any], context_wrapper: RunContextWrapper[Any], config: RunConfig, ) -> list[RunItem]: """Run computer actions serially and emit screenshot outputs.""" from .tool_actions import ComputerAction results: list[RunItem] = [] for action in actions: acknowledged: list[ComputerCallOutputAcknowledgedSafetyCheck] | None = None if action.tool_call.pending_safety_checks and action.computer_tool.on_safety_check: acknowledged = [] for check in action.tool_call.pending_safety_checks: data = ComputerToolSafetyCheckData( ctx_wrapper=context_wrapper, agent=public_agent, tool_call=action.tool_call, safety_check=check, ) maybe = action.computer_tool.on_safety_check(data) ack = await maybe if inspect.isawaitable(maybe) else maybe if ack: acknowledged.append( ComputerCallOutputAcknowledgedSafetyCheck( id=check.id, code=check.code, message=check.message, ) ) else: raise UserError("Computer tool safety check was not acknowledged") results.append( await ComputerAction.execute( agent=public_agent, action=action, hooks=hooks, context_wrapper=context_wrapper, config=config, acknowledged_safety_checks=acknowledged, ) ) return results async def execute_approved_tools( *, agent: Agent[Any], interruptions: list[Any], context_wrapper: RunContextWrapper[Any], generated_items: list[RunItem], run_config: RunConfig, hooks: RunHooks[Any], all_tools: list[Tool] | None = None, ) -> None: """Execute tools that have been approved after an interruption (HITL resume path).""" tool_runs: list[ToolRunFunction] = [] tool_map: dict[NamedToolLookupKey, Tool] = cast( dict[NamedToolLookupKey, Tool], build_function_tool_lookup_map( [tool for tool in all_tools or [] if isinstance(tool, FunctionTool)] ), ) for tool in all_tools or []: if isinstance(tool, FunctionTool): continue if hasattr(tool, "name"): tool_name = getattr(tool, "name", None) if isinstance(tool_name, str) and tool_name: tool_map[tool_name] = tool def _append_error( message: str, *, tool_call: Any, tool_name: str, call_id: str, tool_origin: ToolOrigin | None = None, ) -> None: append_approval_error_output( message=message, tool_call=tool_call, tool_name=tool_name, call_id=call_id, generated_items=generated_items, agent=agent, tool_origin=tool_origin, ) async def _resolve_tool_run( interruption: Any, ) -> tuple[ResponseFunctionToolCall, FunctionTool, str, str] | None: tool_call = interruption.raw_item tool_name = interruption.name or RunContextWrapper._resolve_tool_name(interruption) tool_namespace = getattr(interruption, "tool_namespace", None) tool_lookup_key = getattr( interruption, "tool_lookup_key", None ) or get_function_tool_lookup_key( tool_name, tool_namespace, ) approval_key = tool_lookup_key display_tool_name = tool_trace_name(tool_name, tool_namespace) or tool_name or "unknown" if not tool_name: _append_error( message="Tool approval item missing tool name.", tool_call=tool_call, tool_name="unknown", call_id="unknown", ) return None call_id = extract_tool_call_id(tool_call) if not call_id: resolved_tool = tool_map.get(approval_key) if approval_key is not None else None if resolved_tool is None and tool_namespace is None: resolved_tool = tool_map.get(tool_name) _append_error( message="Tool approval item missing call ID.", tool_call=tool_call, tool_name=tool_name, call_id="unknown", tool_origin=( get_function_tool_origin(resolved_tool) if isinstance(resolved_tool, FunctionTool) else None ), ) return None resolved_tool = tool_map.get(approval_key) if approval_key is not None else None if resolved_tool is None and tool_namespace is None: resolved_tool = tool_map.get(tool_name) approval_status = context_wrapper.get_approval_status( tool_name, call_id, tool_namespace=tool_namespace, existing_pending=interruption, tool_lookup_key=tool_lookup_key, ) if approval_status is False: message = REJECTION_MESSAGE if isinstance(resolved_tool, FunctionTool): message = await resolve_approval_rejection_message( context_wrapper=context_wrapper, run_config=run_config, tool_type="function", tool_name=display_tool_name, call_id=call_id, tool_namespace=tool_namespace, tool_lookup_key=tool_lookup_key, existing_pending=interruption, ) _append_error( message=message, tool_call=tool_call, tool_name=tool_name, call_id=call_id, tool_origin=( get_function_tool_origin(resolved_tool) if isinstance(resolved_tool, FunctionTool) else None ), ) return None if approval_status is not True: _append_error( message="Tool approval status unclear.", tool_call=tool_call, tool_name=tool_name, call_id=call_id, tool_origin=( get_function_tool_origin(resolved_tool) if isinstance(resolved_tool, FunctionTool) else None ), ) return None tool = resolved_tool if tool is None: _append_error( message=f"Tool '{display_tool_name}' not found.", tool_call=tool_call, tool_name=tool_name, call_id=call_id, ) return None if not isinstance(tool, FunctionTool): _append_error( message=f"Tool '{display_tool_name}' is not a function tool.", tool_call=tool_call, tool_name=tool_name, call_id=call_id, ) return None if not isinstance(tool_call, ResponseFunctionToolCall): _append_error( message=( f"Tool '{tool_name}' approval item has invalid raw_item type for execution." ), tool_call=tool_call, tool_name=tool_name, call_id=call_id, ) return None return tool_call, tool, tool_name, call_id for interruption in interruptions: resolved = await _resolve_tool_run(interruption) if resolved is None: continue tool_call, tool, tool_name, _ = resolved tool_runs.append(ToolRunFunction(function_tool=tool, tool_call=tool_call)) if tool_runs: function_results, _, _ = await execute_function_tool_calls( bindings=bind_public_agent(agent), tool_runs=tool_runs, hooks=hooks, context_wrapper=context_wrapper, config=run_config, ) for result in function_results: if isinstance(result.run_item, RunItemBase): generated_items.append(result.run_item) # -------------------------- # Private helpers # -------------------------- async def _execute_tool_input_guardrails( *, func_tool: FunctionTool, tool_context: ToolContext[Any], agent: Agent[Any], tool_input_guardrail_results: list[ToolInputGuardrailResult], ) -> str | None: """Execute input guardrails for a tool call and return a rejection message if any.""" if not func_tool.tool_input_guardrails: return None for guardrail in func_tool.tool_input_guardrails: gr_out = await guardrail.run( ToolInputGuardrailData( context=tool_context, agent=agent, ) ) tool_input_guardrail_results.append( ToolInputGuardrailResult( guardrail=guardrail, output=gr_out, ) ) if gr_out.behavior["type"] == "raise_exception": raise ToolInputGuardrailTripwireTriggered(guardrail=guardrail, output=gr_out) elif gr_out.behavior["type"] == "reject_content": return gr_out.behavior["message"] return None async def _execute_tool_output_guardrails( *, func_tool: FunctionTool, tool_context: ToolContext[Any], agent: Agent[Any], real_result: Any, tool_output_guardrail_results: list[ToolOutputGuardrailResult], ) -> Any: """Execute output guardrails for a tool call and return the final result.""" if not func_tool.tool_output_guardrails: return real_result final_result = real_result for output_guardrail in func_tool.tool_output_guardrails: gr_out = await output_guardrail.run( ToolOutputGuardrailData( context=tool_context, agent=agent, output=real_result, ) ) tool_output_guardrail_results.append( ToolOutputGuardrailResult( guardrail=output_guardrail, output=gr_out, ) ) if gr_out.behavior["type"] == "raise_exception": raise ToolOutputGuardrailTripwireTriggered(guardrail=output_guardrail, output=gr_out) elif gr_out.behavior["type"] == "reject_content": final_result = gr_out.behavior["message"] break return final_result def _normalize_exit_code(value: Any) -> int | None: """Convert arbitrary exit code types into an int if possible.""" if value is None: return None try: return int(value) except (TypeError, ValueError): return None def _is_hosted_mcp_approval_request(raw_item: Any) -> bool: """Detect hosted MCP approval request payloads emitted by the provider.""" if isinstance(raw_item, McpApprovalRequest): return True if not isinstance(raw_item, dict): return False provider_data = raw_item.get("provider_data", {}) return ( raw_item.get("type") == "hosted_tool_call" and provider_data.get("type") == "mcp_approval_request" )