from __future__ import annotations import asyncio import inspect from collections.abc import Awaitable, Callable, Mapping, Sequence from typing import Any, Literal, cast from openai.types.responses import ( ResponseCompactionItem, ResponseComputerToolCall, ResponseCustomToolCall, ResponseFileSearchToolCall, ResponseFunctionShellToolCallOutput, ResponseFunctionToolCall, ResponseFunctionWebSearch, ResponseOutputMessage, ) from openai.types.responses.response_code_interpreter_tool_call import ( ResponseCodeInterpreterToolCall, ) from openai.types.responses.response_output_item import ( ImageGenerationCall, LocalShellCall, McpApprovalRequest, McpCall, McpListTools, ) from openai.types.responses.response_reasoning_item import ResponseReasoningItem from .._mcp_tool_metadata import collect_mcp_list_tools_metadata from .._tool_identity import ( build_function_tool_lookup_map, get_function_tool_lookup_key, get_function_tool_lookup_key_for_call, get_function_tool_lookup_key_for_tool, get_tool_call_namespace, get_tool_call_qualified_name, get_tool_call_trace_name, normalize_tool_call_for_function_tool, should_allow_bare_name_approval_alias, ) from ..agent import Agent, ToolsToFinalOutputResult from ..agent_output import AgentOutputSchemaBase from ..agent_tool_state import get_agent_tool_state_scope, peek_agent_tool_run_result from ..exceptions import ModelBehaviorError, UserError from ..handoffs import Handoff, HandoffInputData, HandoffInputFilter, nest_handoff_history from ..items import ( CompactionItem, HandoffCallItem, HandoffOutputItem, ItemHelpers, MCPApprovalRequestItem, MCPListToolsItem, MessageOutputItem, ModelResponse, ReasoningItem, RunItem, ToolApprovalItem, ToolCallItem, ToolCallOutputItem, ToolSearchCallItem, ToolSearchOutputItem, TResponseInputItem, coerce_tool_search_call_raw_item, coerce_tool_search_output_raw_item, ) from ..lifecycle import RunHooks from ..logger import logger from ..run_config import RunConfig from ..run_context import AgentHookContext, RunContextWrapper, TContext from ..run_state import RunState from ..stream_events import StreamEvent from ..tool import ( ApplyPatchTool, ComputerTool, CustomTool, FunctionTool, FunctionToolResult, HostedMCPTool, LocalShellTool, ShellTool, Tool, ToolOrigin, ToolOriginType, get_function_tool_origin, ) from ..tool_guardrails import ToolInputGuardrailResult, ToolOutputGuardrailResult from ..tracing import SpanError, handoff_span from ..util import _coro, _error_tracing from ..util._approvals import evaluate_needs_approval_setting from .agent_bindings import AgentBindings from .items import ( REJECTION_MESSAGE, apply_patch_rejection_item, function_rejection_item, shell_rejection_item, ) from .run_steps import ( NOT_FINAL_OUTPUT, NextStepFinalOutput, NextStepHandoff, NextStepInterruption, NextStepRunAgain, ProcessedResponse, QueueCompleteSentinel, SingleStepResult, ToolRunApplyPatchCall, ToolRunComputerAction, ToolRunCustom, ToolRunFunction, ToolRunHandoff, ToolRunLocalShellCall, ToolRunMCPApprovalRequest, ToolRunShellCall, ) from .streaming import stream_step_items_to_queue from .tool_execution import ( build_litellm_json_tool_call, coerce_apply_patch_operations, coerce_shell_call, extract_apply_patch_call_id, extract_shell_call_id, extract_tool_call_id, function_needs_approval, get_mapping_or_attr, index_approval_items_by_call_id, is_apply_patch_name, parse_apply_patch_custom_input, parse_apply_patch_function_args, process_hosted_mcp_approvals, resolve_approval_rejection_message, resolve_enabled_function_tools, should_keep_hosted_mcp_item, ) from .tool_planning import ( _append_mcp_callback_results, _build_plan_for_fresh_turn, _build_plan_for_resume_turn, _build_tool_output_index, _build_tool_result_items, _collect_runs_by_approval, _collect_tool_interruptions, _dedupe_tool_call_items, _execute_tool_plan, _make_unique_item_appender, _select_function_tool_runs_for_resume, ) __all__ = [ "execute_final_output_step", "execute_final_output", "execute_handoffs", "check_for_final_output_from_tools", "process_model_response", "execute_tools_and_side_effects", "resolve_interrupted_turn", "get_single_step_result_from_response", "run_final_output_hooks", ] async def _maybe_finalize_from_tool_results( *, public_agent: Agent[TContext], original_input: str | list[TResponseInputItem], new_response: ModelResponse, pre_step_items: list[RunItem], new_step_items: list[RunItem], function_results: list[FunctionToolResult], hooks: RunHooks[TContext], context_wrapper: RunContextWrapper[TContext], tool_input_guardrail_results: list[ToolInputGuardrailResult], tool_output_guardrail_results: list[ToolOutputGuardrailResult], ) -> SingleStepResult | None: check_tool_use = await check_for_final_output_from_tools( public_agent, function_results, context_wrapper ) if not check_tool_use.is_final_output: return None if not public_agent.output_type or public_agent.output_type is str: check_tool_use.final_output = str(check_tool_use.final_output) if check_tool_use.final_output is None: logger.error( "Model returned a final output of None. Not raising an error because we assume" "you know what you're doing." ) return await execute_final_output( public_agent=public_agent, original_input=original_input, new_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, final_output=check_tool_use.final_output, hooks=hooks, context_wrapper=context_wrapper, tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, ) async def run_final_output_hooks( agent: Agent[TContext], hooks: RunHooks[TContext], context_wrapper: RunContextWrapper[TContext], final_output: Any, ) -> None: agent_hook_context = AgentHookContext( context=context_wrapper.context, usage=context_wrapper.usage, _approvals=context_wrapper._approvals, turn_input=context_wrapper.turn_input, ) await asyncio.gather( hooks.on_agent_end(agent_hook_context, agent, final_output), agent.hooks.on_end(agent_hook_context, agent, final_output) if agent.hooks else _coro.noop_coroutine(), ) async def execute_final_output_step( *, public_agent: Agent[Any], original_input: str | list[TResponseInputItem], new_response: ModelResponse, pre_step_items: list[RunItem], new_step_items: list[RunItem], final_output: Any, hooks: RunHooks[Any], context_wrapper: RunContextWrapper[Any], tool_input_guardrail_results: list[ToolInputGuardrailResult], tool_output_guardrail_results: list[ToolOutputGuardrailResult], run_final_output_hooks_fn: Callable[ [Agent[Any], RunHooks[Any], RunContextWrapper[Any], Any], Awaitable[None] ] | None = None, ) -> SingleStepResult: """Finalize a turn once final output is known and run end hooks.""" final_output_hooks = run_final_output_hooks_fn or run_final_output_hooks await final_output_hooks(public_agent, hooks, context_wrapper, final_output) return SingleStepResult( original_input=original_input, model_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, next_step=NextStepFinalOutput(final_output), tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, output_guardrail_results=[], ) async def execute_final_output( *, public_agent: Agent[Any], original_input: str | list[TResponseInputItem], new_response: ModelResponse, pre_step_items: list[RunItem], new_step_items: list[RunItem], final_output: Any, hooks: RunHooks[Any], context_wrapper: RunContextWrapper[Any], tool_input_guardrail_results: list[ToolInputGuardrailResult], tool_output_guardrail_results: list[ToolOutputGuardrailResult], run_final_output_hooks_fn: Callable[ [Agent[Any], RunHooks[Any], RunContextWrapper[Any], Any], Awaitable[None] ] | None = None, ) -> SingleStepResult: """Convenience wrapper to finalize a turn and run end hooks.""" return await execute_final_output_step( public_agent=public_agent, original_input=original_input, new_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, final_output=final_output, hooks=hooks, context_wrapper=context_wrapper, tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, run_final_output_hooks_fn=run_final_output_hooks_fn, ) def _resolve_server_managed_handoff_behavior( *, handoff: Handoff[Any, Agent[Any]], from_agent: Agent[Any], to_agent: Agent[Any], run_config: RunConfig, server_manages_conversation: bool, input_filter: HandoffInputFilter | None, should_nest_history: bool, ) -> tuple[HandoffInputFilter | None, bool]: if not server_manages_conversation: return input_filter, should_nest_history if input_filter is not None: raise UserError( "Server-managed conversations do not support handoff input filters. " "Remove Handoff.input_filter or RunConfig.handoff_input_filter, " "or disable conversation_id, previous_response_id, and auto_previous_response_id." ) if not should_nest_history: return input_filter, should_nest_history logger.warning( "Server-managed conversations do not support nest_handoff_history for handoff " "%s -> %s. Disabling nested handoff history and continuing with delta-only input.", from_agent.name, to_agent.name, ) return input_filter, False async def execute_handoffs( *, public_agent: Agent[TContext], original_input: str | list[TResponseInputItem], pre_step_items: list[RunItem], new_step_items: list[RunItem], new_response: ModelResponse, run_handoffs: list[ToolRunHandoff], hooks: RunHooks[TContext], context_wrapper: RunContextWrapper[TContext], run_config: RunConfig, server_manages_conversation: bool = False, nest_handoff_history_fn: Callable[..., HandoffInputData] | None = None, ) -> SingleStepResult: """Execute a handoff and prepare the next turn for the new agent.""" def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffInputData: if nest_handoff_history_fn is None: return nest_handoff_history(data, history_mapper=mapper) return nest_handoff_history_fn(data, mapper) multiple_handoffs = len(run_handoffs) > 1 if multiple_handoffs: output_message = "Multiple handoffs detected, ignoring this one." new_step_items.extend( [ ToolCallOutputItem( output=output_message, raw_item=ItemHelpers.tool_call_output_item(handoff.tool_call, output_message), agent=public_agent, ) for handoff in run_handoffs[1:] ] ) actual_handoff = run_handoffs[0] with handoff_span(from_agent=public_agent.name) as span_handoff: handoff = actual_handoff.handoff new_agent: Agent[Any] = await handoff.on_invoke_handoff( context_wrapper, actual_handoff.tool_call.arguments ) span_handoff.span_data.to_agent = new_agent.name if multiple_handoffs: requested_agents = [handoff.handoff.agent_name for handoff in run_handoffs] span_handoff.set_error( SpanError( message="Multiple handoffs requested", data={ "requested_agents": requested_agents, }, ) ) new_step_items.append( HandoffOutputItem( agent=public_agent, raw_item=ItemHelpers.tool_call_output_item( actual_handoff.tool_call, handoff.get_transfer_message(new_agent), ), source_agent=public_agent, target_agent=new_agent, ) ) await asyncio.gather( hooks.on_handoff( context=context_wrapper, from_agent=public_agent, to_agent=new_agent, ), ( public_agent.hooks.on_handoff( context_wrapper, agent=new_agent, source=public_agent, ) if public_agent.hooks else _coro.noop_coroutine() ), ) input_filter = handoff.input_filter or ( run_config.handoff_input_filter if run_config else None ) handoff_nest_setting = handoff.nest_handoff_history should_nest_history = ( handoff_nest_setting if handoff_nest_setting is not None else run_config.nest_handoff_history ) input_filter, should_nest_history = _resolve_server_managed_handoff_behavior( handoff=handoff, from_agent=public_agent, to_agent=new_agent, run_config=run_config, server_manages_conversation=server_manages_conversation, input_filter=input_filter, should_nest_history=should_nest_history, ) handoff_input_data: HandoffInputData | None = None session_step_items: list[RunItem] | None = None if input_filter or should_nest_history: handoff_input_data = HandoffInputData( input_history=tuple(original_input) if isinstance(original_input, list) else original_input, pre_handoff_items=tuple(pre_step_items), new_items=tuple(new_step_items), run_context=context_wrapper, ) if input_filter and handoff_input_data is not None: filter_name = getattr(input_filter, "__qualname__", repr(input_filter)) from_agent = getattr(public_agent, "name", public_agent.__class__.__name__) to_agent = getattr(new_agent, "name", new_agent.__class__.__name__) logger.debug( "Filtering handoff inputs with %s for %s -> %s", filter_name, from_agent, to_agent, ) if not callable(input_filter): _error_tracing.attach_error_to_span( span_handoff, SpanError( message="Invalid input filter", data={"details": "not callable()"}, ), ) raise UserError(f"Invalid input filter: {input_filter}") filtered = input_filter(handoff_input_data) if inspect.isawaitable(filtered): filtered = await filtered if not isinstance(filtered, HandoffInputData): _error_tracing.attach_error_to_span( span_handoff, SpanError( message="Invalid input filter result", data={"details": "not a HandoffInputData"}, ), ) raise UserError(f"Invalid input filter result: {filtered}") original_input = ( filtered.input_history if isinstance(filtered.input_history, str) else list(filtered.input_history) ) pre_step_items = list(filtered.pre_handoff_items) new_step_items = list(filtered.new_items) # For custom input filters, keep full new_items for session history and # use input_items for model input when provided. if filtered.input_items is not None: session_step_items = list(filtered.new_items) new_step_items = list(filtered.input_items) else: session_step_items = None elif should_nest_history and handoff_input_data is not None: nested = nest_history(handoff_input_data, run_config.handoff_history_mapper) original_input = ( nested.input_history if isinstance(nested.input_history, str) else list(nested.input_history) ) pre_step_items = list(nested.pre_handoff_items) # Keep full new_items for session history. session_step_items = list(nested.new_items) # Use input_items (filtered) for model input if available. if nested.input_items is not None: new_step_items = list(nested.input_items) else: new_step_items = session_step_items else: # No filtering or nesting - session_step_items not needed. session_step_items = None return SingleStepResult( original_input=original_input, model_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, next_step=NextStepHandoff(new_agent), tool_input_guardrail_results=[], tool_output_guardrail_results=[], session_step_items=session_step_items, ) async def check_for_final_output_from_tools( agent: Agent[TContext], tool_results: list[FunctionToolResult], context_wrapper: RunContextWrapper[TContext], ) -> ToolsToFinalOutputResult: """Determine if tool results should produce a final output.""" if not tool_results: return NOT_FINAL_OUTPUT if agent.tool_use_behavior == "run_llm_again": return NOT_FINAL_OUTPUT elif agent.tool_use_behavior == "stop_on_first_tool": return ToolsToFinalOutputResult(is_final_output=True, final_output=tool_results[0].output) elif isinstance(agent.tool_use_behavior, dict): names = agent.tool_use_behavior.get("stop_at_tool_names", []) for tool_result in tool_results: if tool_result.tool.name in names or tool_result.tool.qualified_name in names: return ToolsToFinalOutputResult( is_final_output=True, final_output=tool_result.output ) return ToolsToFinalOutputResult(is_final_output=False, final_output=None) elif callable(agent.tool_use_behavior): if inspect.iscoroutinefunction(agent.tool_use_behavior): return await cast( Awaitable[ToolsToFinalOutputResult], agent.tool_use_behavior(context_wrapper, tool_results), ) return cast( ToolsToFinalOutputResult, agent.tool_use_behavior(context_wrapper, tool_results) ) logger.error("Invalid tool_use_behavior: %s", agent.tool_use_behavior) raise UserError(f"Invalid tool_use_behavior: {agent.tool_use_behavior}") async def execute_tools_and_side_effects( *, bindings: AgentBindings[TContext], original_input: str | list[TResponseInputItem], pre_step_items: list[RunItem], new_response: ModelResponse, processed_response: ProcessedResponse, output_schema: AgentOutputSchemaBase | None, hooks: RunHooks[TContext], context_wrapper: RunContextWrapper[TContext], run_config: RunConfig, server_manages_conversation: bool = False, ) -> SingleStepResult: """Run one turn of the loop, coordinating tools, approvals, guardrails, and handoffs.""" public_agent = bindings.public_agent execute_final_output_call = execute_final_output execute_handoffs_call = execute_handoffs pre_step_items = list(pre_step_items) approval_items_by_call_id = index_approval_items_by_call_id(pre_step_items) plan = _build_plan_for_fresh_turn( processed_response=processed_response, agent=public_agent, context_wrapper=context_wrapper, approval_items_by_call_id=approval_items_by_call_id, ) new_step_items = _dedupe_tool_call_items( existing_items=pre_step_items, new_items=processed_response.new_items, ) ( function_results, tool_input_guardrail_results, tool_output_guardrail_results, computer_results, custom_tool_results, shell_results, apply_patch_results, local_shell_results, ) = await _execute_tool_plan( plan=plan, bindings=bindings, hooks=hooks, context_wrapper=context_wrapper, run_config=run_config, ) new_step_items.extend( _build_tool_result_items( function_results=function_results, computer_results=computer_results, custom_tool_results=custom_tool_results, shell_results=shell_results, apply_patch_results=apply_patch_results, local_shell_results=local_shell_results, ) ) interruptions = _collect_tool_interruptions( function_results=function_results, custom_tool_results=custom_tool_results, shell_results=shell_results, apply_patch_results=apply_patch_results, ) if plan.approved_mcp_responses: new_step_items.extend(plan.approved_mcp_responses) if plan.pending_interruptions: interruptions.extend(plan.pending_interruptions) new_step_items.extend(plan.pending_interruptions) processed_response.interruptions = interruptions if interruptions: return SingleStepResult( original_input=original_input, model_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, next_step=NextStepInterruption(interruptions=interruptions), tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, processed_response=processed_response, ) await _append_mcp_callback_results( agent=public_agent, requests=plan.mcp_requests_with_callback, context_wrapper=context_wrapper, append_item=new_step_items.append, ) if run_handoffs := processed_response.handoffs: return await execute_handoffs_call( public_agent=public_agent, original_input=original_input, pre_step_items=pre_step_items, new_step_items=new_step_items, new_response=new_response, run_handoffs=run_handoffs, hooks=hooks, context_wrapper=context_wrapper, run_config=run_config, server_manages_conversation=server_manages_conversation, ) tool_final_output = await _maybe_finalize_from_tool_results( public_agent=public_agent, original_input=original_input, new_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, function_results=function_results, hooks=hooks, context_wrapper=context_wrapper, tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, ) if tool_final_output is not None: return tool_final_output message_items = [item for item in new_step_items if isinstance(item, MessageOutputItem)] potential_final_output_text = ( ItemHelpers.extract_text(message_items[-1].raw_item) if message_items else None ) if not processed_response.has_tools_or_approvals_to_run(): has_tool_activity_without_message = not message_items and bool( processed_response.tools_used ) if not has_tool_activity_without_message: if output_schema and not output_schema.is_plain_text() and potential_final_output_text: final_output = output_schema.validate_json(potential_final_output_text) return await execute_final_output_call( public_agent=public_agent, original_input=original_input, new_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, final_output=final_output, hooks=hooks, context_wrapper=context_wrapper, tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, ) if not output_schema or output_schema.is_plain_text(): return await execute_final_output_call( public_agent=public_agent, original_input=original_input, new_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, final_output=potential_final_output_text or "", hooks=hooks, context_wrapper=context_wrapper, tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, ) return SingleStepResult( original_input=original_input, model_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, next_step=NextStepRunAgain(), tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, ) async def resolve_interrupted_turn( *, bindings: AgentBindings[TContext], original_input: str | list[TResponseInputItem], original_pre_step_items: list[RunItem], new_response: ModelResponse, processed_response: ProcessedResponse, hooks: RunHooks[TContext], context_wrapper: RunContextWrapper[TContext], run_config: RunConfig, server_manages_conversation: bool = False, run_state: RunState | None = None, nest_handoff_history_fn: Callable[..., HandoffInputData] | None = None, ) -> SingleStepResult: """Continue a turn that was previously interrupted waiting for tool approval.""" public_agent = bindings.public_agent execution_agent = bindings.execution_agent execute_handoffs_call = execute_handoffs def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffInputData: if nest_handoff_history_fn is None: return nest_handoff_history(data, history_mapper=mapper) return nest_handoff_history_fn(data, mapper) def _pending_approvals_from_state() -> list[ToolApprovalItem]: if ( run_state is not None and hasattr(run_state, "_current_step") and isinstance(run_state._current_step, NextStepInterruption) ): return [ item for item in run_state._current_step.interruptions if isinstance(item, ToolApprovalItem) ] return [item for item in original_pre_step_items if isinstance(item, ToolApprovalItem)] async def _record_function_rejection( call_id: str | None, tool_call: ResponseFunctionToolCall, function_tool: FunctionTool, ) -> None: if isinstance(call_id, str) and call_id in rejected_function_call_ids: return rejection_message = REJECTION_MESSAGE if call_id: tool_namespace = get_tool_call_namespace(tool_call) rejection_message = await resolve_approval_rejection_message( context_wrapper=context_wrapper, run_config=run_config, tool_type="function", tool_name=get_tool_call_trace_name(tool_call) or function_tool.name, call_id=call_id, tool_namespace=tool_namespace, tool_lookup_key=get_function_tool_lookup_key_for_tool(function_tool), existing_pending=approval_items_by_call_id.get(call_id), ) rejected_function_outputs.append( function_rejection_item( public_agent, tool_call, rejection_message=rejection_message, scope_id=tool_state_scope_id, tool_origin=get_function_tool_origin(function_tool), ) ) if isinstance(call_id, str): rejected_function_call_ids.add(call_id) async def _function_requires_approval(run: ToolRunFunction) -> bool: call_id = run.tool_call.call_id if call_id and call_id in approval_items_by_call_id: return True try: return await function_needs_approval( run.function_tool, context_wrapper, run.tool_call, ) except UserError: raise except Exception: return True try: context_wrapper.turn_input = ItemHelpers.input_to_new_input_list(original_input) except Exception: context_wrapper.turn_input = [] pending_approval_items = _pending_approvals_from_state() approval_items_by_call_id = index_approval_items_by_call_id(pending_approval_items) tool_state_scope_id = get_agent_tool_state_scope(context_wrapper) rejected_function_outputs: list[RunItem] = [] rejected_function_call_ids: set[str] = set() rerun_function_call_ids: set[str] = set() pending_interruptions: list[ToolApprovalItem] = [] pending_interruption_keys: set[str] = set() output_index = _build_tool_output_index(original_pre_step_items) def _has_output_item(call_id: str, expected_type: str) -> bool: return (expected_type, call_id) in output_index def _shell_call_id_from_run(run: ToolRunShellCall) -> str: return extract_shell_call_id(run.tool_call) def _apply_patch_call_id_from_run(run: ToolRunApplyPatchCall) -> str: return extract_apply_patch_call_id(run.tool_call) def _custom_call_id_from_run(run: ToolRunCustom) -> str: call_id = extract_tool_call_id(run.tool_call) if not call_id: raise ModelBehaviorError("Custom tool call is missing call_id.") return call_id def _computer_call_id_from_run(run: ToolRunComputerAction) -> str: call_id = extract_tool_call_id(run.tool_call) if not call_id: raise ModelBehaviorError("Computer action is missing call_id.") return call_id def _shell_tool_name(run: ToolRunShellCall) -> str: return run.shell_tool.name def _apply_patch_tool_name(run: ToolRunApplyPatchCall) -> str: return run.apply_patch_tool.name def _custom_tool_name(run: ToolRunCustom) -> str: return run.custom_tool.name async def _build_shell_rejection(run: ToolRunShellCall, call_id: str) -> RunItem: rejection_message = await resolve_approval_rejection_message( context_wrapper=context_wrapper, run_config=run_config, tool_type="shell", tool_name=run.shell_tool.name, call_id=call_id, ) return cast( RunItem, shell_rejection_item( public_agent, call_id, rejection_message=rejection_message, ), ) async def _build_apply_patch_rejection(run: ToolRunApplyPatchCall, call_id: str) -> RunItem: rejection_message = await resolve_approval_rejection_message( context_wrapper=context_wrapper, run_config=run_config, tool_type="apply_patch", tool_name=run.apply_patch_tool.name, call_id=call_id, ) return cast( RunItem, apply_patch_rejection_item( public_agent, call_id, output_type="apply_patch_call_output", rejection_message=rejection_message, ), ) async def _build_custom_rejection(run: ToolRunCustom, call_id: str) -> RunItem: rejection_message = await resolve_approval_rejection_message( context_wrapper=context_wrapper, run_config=run_config, tool_type="custom", tool_name=run.custom_tool.name, call_id=call_id, ) return ToolCallOutputItem( agent=public_agent, output=rejection_message, raw_item=cast( Any, { "type": "custom_tool_call_output", "call_id": call_id, "output": rejection_message, }, ), ) async def _shell_needs_approval(run: ToolRunShellCall) -> bool: shell_call = coerce_shell_call(run.tool_call) return await evaluate_needs_approval_setting( run.shell_tool.needs_approval, context_wrapper, shell_call.action, shell_call.call_id, ) async def _apply_patch_needs_approval(run: ToolRunApplyPatchCall) -> bool: operations = coerce_apply_patch_operations( run.tool_call, context_wrapper=context_wrapper, ) call_id = extract_apply_patch_call_id(run.tool_call) for operation in operations: if await evaluate_needs_approval_setting( run.apply_patch_tool.needs_approval, context_wrapper, operation, call_id ): return True return False async def _custom_tool_needs_approval(run: ToolRunCustom) -> bool: tool_input = get_mapping_or_attr(run.tool_call, "input") call_id = _custom_call_id_from_run(run) if not isinstance(tool_input, str): raise ModelBehaviorError("Custom tool call is missing input.") return await evaluate_needs_approval_setting( run.custom_tool.runtime_needs_approval(), context_wrapper, tool_input, call_id, ) def _shell_output_exists(call_id: str) -> bool: return _has_output_item(call_id, "shell_call_output") def _apply_patch_output_exists(call_id: str) -> bool: return _has_output_item(call_id, "apply_patch_call_output") def _custom_tool_output_exists(call_id: str) -> bool: return _has_output_item(call_id, "custom_tool_call_output") def _computer_output_exists(call_id: str) -> bool: return _has_output_item(call_id, "computer_call_output") def _nested_interruptions_status( interruptions: Sequence[ToolApprovalItem], ) -> Literal["approved", "pending", "rejected"]: has_pending = False for interruption in interruptions: call_id = extract_tool_call_id(interruption.raw_item) if not call_id: has_pending = True continue status = context_wrapper.get_approval_status( interruption.tool_name or "", call_id, tool_namespace=interruption.tool_namespace, existing_pending=interruption, ) if status is False: return "rejected" if status is None: has_pending = True return "pending" if has_pending else "approved" def _function_output_exists(run: ToolRunFunction) -> bool: call_id = extract_tool_call_id(run.tool_call) if not call_id: return False pending_run_result = peek_agent_tool_run_result( run.tool_call, scope_id=tool_state_scope_id, ) if pending_run_result and getattr(pending_run_result, "interruptions", None): status = _nested_interruptions_status(pending_run_result.interruptions) if status in ("approved", "rejected"): rerun_function_call_ids.add(call_id) return False return True return _has_output_item(call_id, "function_call_output") def _add_pending_interruption(item: ToolApprovalItem | None) -> None: if item is None: return call_id = extract_tool_call_id(item.raw_item) key = call_id or f"raw:{id(item.raw_item)}" if key in pending_interruption_keys: return pending_interruption_keys.add(key) pending_interruptions.append(item) def _allow_legacy_name_agent_match() -> bool: schema_version = getattr(run_state, "_schema_version", None) if not isinstance(schema_version, str): return False try: version_parts = tuple(int(part) for part in schema_version.split(".")) except ValueError: return False # Schema 1.6 and earlier only serialized approval owners by agent name. With duplicate-name # agents, deserialization can legitimately resolve the approval to a sibling instance, so # resume must accept a same-name match for those legacy snapshots. Schema 1.7+ persists # duplicate-name identities, so newer snapshots should continue requiring object identity. return version_parts < (1, 7) allow_legacy_name_agent_match = _allow_legacy_name_agent_match() def _approval_matches_agent(approval: ToolApprovalItem) -> bool: approval_agent = approval.agent if approval_agent is None: return False if approval_agent is public_agent: return True return allow_legacy_name_agent_match and approval_agent.name == public_agent.name available_function_tools = await resolve_enabled_function_tools( execution_agent, context_wrapper, ) approval_rebuild_function_tools = available_function_tools if pending_approval_items and execution_agent.mcp_servers: approval_rebuild_function_tools = [ tool for tool in await execution_agent.get_all_tools(context_wrapper) if isinstance(tool, FunctionTool) ] async def _rebuild_function_runs_from_approvals() -> list[ToolRunFunction]: if not pending_approval_items: return [] tool_map = build_function_tool_lookup_map(approval_rebuild_function_tools) existing_pending_call_ids: set[str] = set() for existing_pending in pending_interruptions: if isinstance(existing_pending, ToolApprovalItem): existing_call_id = extract_tool_call_id(existing_pending.raw_item) if existing_call_id: existing_pending_call_ids.add(existing_call_id) rebuilt_runs: list[ToolRunFunction] = [] def _add_unmatched_pending(approval: ToolApprovalItem) -> None: call_id = extract_tool_call_id(approval.raw_item) if not call_id: _add_pending_interruption(approval) return tool_name = approval.tool_name or "" approval_status = context_wrapper.get_approval_status( tool_name, call_id, tool_namespace=approval.tool_namespace, existing_pending=approval, ) if approval_status is None: _add_pending_interruption(approval) for approval in pending_approval_items: if not isinstance(approval, ToolApprovalItem): continue if not _approval_matches_agent(approval): _add_unmatched_pending(approval) continue raw = approval.raw_item raw_type = get_mapping_or_attr(raw, "type") if raw_type != "function_call": _add_unmatched_pending(approval) continue name = get_mapping_or_attr(raw, "name") namespace = get_tool_call_namespace(raw) if namespace is None and isinstance(approval.tool_namespace, str): namespace = approval.tool_namespace approval_key = getattr(approval, "tool_lookup_key", None) if approval_key is None: approval_key = get_function_tool_lookup_key(name, namespace) resolved_tool = tool_map.get(approval_key) if approval_key is not None else None if not (isinstance(name, str) and resolved_tool is not None): _add_unmatched_pending(approval) continue rebuilt_call_id: str | None arguments: str | None tool_call: ResponseFunctionToolCall if isinstance(raw, ResponseFunctionToolCall): rebuilt_call_id = raw.call_id arguments = raw.arguments tool_call = raw else: rebuilt_call_id = extract_tool_call_id(raw) arguments = get_mapping_or_attr(raw, "arguments") or "{}" status = get_mapping_or_attr(raw, "status") if not (isinstance(rebuilt_call_id, str) and isinstance(arguments, str)): _add_unmatched_pending(approval) continue valid_status: Literal["in_progress", "completed", "incomplete"] | None = None if isinstance(status, str) and status in ( "in_progress", "completed", "incomplete", ): valid_status = status # type: ignore[assignment] tool_call_payload: dict[str, Any] = { "type": "function_call", "name": name, "call_id": rebuilt_call_id, "arguments": arguments, "status": valid_status, } if namespace is not None: tool_call_payload["namespace"] = namespace tool_call = ResponseFunctionToolCall(**tool_call_payload) tool_call = cast( ResponseFunctionToolCall, normalize_tool_call_for_function_tool(tool_call, resolved_tool), ) if not (isinstance(rebuilt_call_id, str) and isinstance(arguments, str)): _add_unmatched_pending(approval) continue approval_status = context_wrapper.get_approval_status( name, rebuilt_call_id, tool_namespace=namespace, existing_pending=approval, ) if approval_status is False: await _record_function_rejection( rebuilt_call_id, tool_call, resolved_tool, ) continue if approval_status is None: if rebuilt_call_id not in existing_pending_call_ids: _add_pending_interruption(approval) existing_pending_call_ids.add(rebuilt_call_id) continue rebuilt_runs.append(ToolRunFunction(function_tool=resolved_tool, tool_call=tool_call)) return rebuilt_runs function_tool_runs = await _select_function_tool_runs_for_resume( processed_response.functions, approval_items_by_call_id=approval_items_by_call_id, context_wrapper=context_wrapper, needs_approval_checker=_function_requires_approval, output_exists_checker=_function_output_exists, record_rejection=_record_function_rejection, pending_interruption_adder=_add_pending_interruption, pending_item_builder=lambda run: ToolApprovalItem( agent=public_agent, raw_item=run.tool_call, tool_name=run.function_tool.name, tool_namespace=get_tool_call_namespace(run.tool_call), tool_origin=get_function_tool_origin(run.function_tool), tool_lookup_key=get_function_tool_lookup_key_for_call(run.tool_call), _allow_bare_name_alias=should_allow_bare_name_approval_alias( run.function_tool, available_function_tools, ), ), ) rebuilt_function_tool_runs = await _rebuild_function_runs_from_approvals() if rebuilt_function_tool_runs: existing_call_ids: set[str] = set() for run in function_tool_runs: call_id = extract_tool_call_id(run.tool_call) if call_id: existing_call_ids.add(call_id) for run in rebuilt_function_tool_runs: call_id = extract_tool_call_id(run.tool_call) if call_id and call_id in existing_call_ids: continue function_tool_runs.append(run) if call_id: existing_call_ids.add(call_id) pending_computer_actions: list[ToolRunComputerAction] = [] for action in processed_response.computer_actions: call_id = _computer_call_id_from_run(action) if _computer_output_exists(call_id): continue pending_computer_actions.append(action) approved_shell_calls, rejected_shell_results = await _collect_runs_by_approval( processed_response.shell_calls, call_id_extractor=_shell_call_id_from_run, tool_name_resolver=_shell_tool_name, rejection_builder=_build_shell_rejection, context_wrapper=context_wrapper, approval_items_by_call_id=approval_items_by_call_id, agent=public_agent, pending_interruption_adder=_add_pending_interruption, needs_approval_checker=_shell_needs_approval, output_exists_checker=_shell_output_exists, ) approved_apply_patch_calls, rejected_apply_patch_results = await _collect_runs_by_approval( processed_response.apply_patch_calls, call_id_extractor=_apply_patch_call_id_from_run, tool_name_resolver=_apply_patch_tool_name, rejection_builder=_build_apply_patch_rejection, context_wrapper=context_wrapper, approval_items_by_call_id=approval_items_by_call_id, agent=public_agent, pending_interruption_adder=_add_pending_interruption, needs_approval_checker=_apply_patch_needs_approval, output_exists_checker=_apply_patch_output_exists, ) approved_custom_tool_calls, rejected_custom_tool_results = await _collect_runs_by_approval( processed_response.custom_tool_calls, call_id_extractor=_custom_call_id_from_run, tool_name_resolver=_custom_tool_name, rejection_builder=_build_custom_rejection, context_wrapper=context_wrapper, approval_items_by_call_id=approval_items_by_call_id, agent=public_agent, pending_interruption_adder=_add_pending_interruption, needs_approval_checker=_custom_tool_needs_approval, output_exists_checker=_custom_tool_output_exists, ) plan = _build_plan_for_resume_turn( processed_response=processed_response, agent=public_agent, context_wrapper=context_wrapper, approval_items_by_call_id=approval_items_by_call_id, pending_interruptions=pending_interruptions, pending_interruption_adder=_add_pending_interruption, function_runs=function_tool_runs, computer_actions=pending_computer_actions, custom_tool_calls=approved_custom_tool_calls, shell_calls=approved_shell_calls, apply_patch_calls=approved_apply_patch_calls, ) ( function_results, tool_input_guardrail_results, tool_output_guardrail_results, computer_results, custom_tool_results, shell_results, apply_patch_results, _local_shell_results, ) = await _execute_tool_plan( plan=plan, bindings=bindings, hooks=hooks, context_wrapper=context_wrapper, run_config=run_config, ) for interruption in _collect_tool_interruptions( function_results=function_results, custom_tool_results=custom_tool_results, shell_results=[], apply_patch_results=[], ): _add_pending_interruption(interruption) new_items, append_if_new = _make_unique_item_appender(original_pre_step_items) for item in _build_tool_result_items( function_results=function_results, computer_results=computer_results, custom_tool_results=custom_tool_results, shell_results=shell_results, apply_patch_results=apply_patch_results, local_shell_results=[], ): append_if_new(item) for rejection_item in rejected_function_outputs: append_if_new(rejection_item) for pending_item in pending_interruptions: if pending_item: append_if_new(pending_item) for shell_rejection in rejected_shell_results: append_if_new(shell_rejection) for custom_tool_rejection in rejected_custom_tool_results: append_if_new(custom_tool_rejection) for apply_patch_rejection in rejected_apply_patch_results: append_if_new(apply_patch_rejection) for approved_response in plan.approved_mcp_responses: append_if_new(approved_response) processed_response.interruptions = pending_interruptions if pending_interruptions: return SingleStepResult( original_input=original_input, model_response=new_response, pre_step_items=original_pre_step_items, new_step_items=new_items, next_step=NextStepInterruption( interruptions=[item for item in pending_interruptions if item] ), tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, processed_response=processed_response, ) await _append_mcp_callback_results( agent=public_agent, requests=plan.mcp_requests_with_callback, context_wrapper=context_wrapper, append_item=append_if_new, ) ( pending_hosted_mcp_approvals, pending_hosted_mcp_approval_ids, ) = process_hosted_mcp_approvals( original_pre_step_items=original_pre_step_items, mcp_approval_requests=processed_response.mcp_approval_requests, context_wrapper=context_wrapper, agent=public_agent, append_item=append_if_new, ) pre_step_items = [ item for item in original_pre_step_items if should_keep_hosted_mcp_item( item, pending_hosted_mcp_approvals=pending_hosted_mcp_approvals, pending_hosted_mcp_approval_ids=pending_hosted_mcp_approval_ids, ) ] if rejected_function_call_ids: pre_step_items = [ item for item in pre_step_items if not ( item.type == "tool_call_output_item" and ( extract_tool_call_id(getattr(item, "raw_item", None)) in rejected_function_call_ids ) ) ] if rerun_function_call_ids: pre_step_items = [ item for item in pre_step_items if not ( item.type == "tool_call_output_item" and ( extract_tool_call_id(getattr(item, "raw_item", None)) in rerun_function_call_ids ) ) ] executed_handoff_call_ids: set[str] = set() for item in original_pre_step_items: if isinstance(item, HandoffCallItem): handoff_call_id = extract_tool_call_id(item.raw_item) if handoff_call_id: executed_handoff_call_ids.add(handoff_call_id) pending_handoffs = [ handoff for handoff in processed_response.handoffs if not handoff.tool_call.call_id or handoff.tool_call.call_id not in executed_handoff_call_ids ] if pending_handoffs: return await execute_handoffs_call( public_agent=public_agent, original_input=original_input, pre_step_items=pre_step_items, new_step_items=new_items, new_response=new_response, run_handoffs=pending_handoffs, hooks=hooks, context_wrapper=context_wrapper, run_config=run_config, server_manages_conversation=server_manages_conversation, nest_handoff_history_fn=nest_history, ) tool_final_output = await _maybe_finalize_from_tool_results( public_agent=public_agent, original_input=original_input, new_response=new_response, pre_step_items=pre_step_items, new_step_items=new_items, function_results=function_results, hooks=hooks, context_wrapper=context_wrapper, tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, ) if tool_final_output is not None: return tool_final_output return SingleStepResult( original_input=original_input, model_response=new_response, pre_step_items=pre_step_items, new_step_items=new_items, next_step=NextStepRunAgain(), tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, ) def process_model_response( *, agent: Agent[Any], all_tools: list[Tool], response: ModelResponse, output_schema: AgentOutputSchemaBase | None, handoffs: list[Handoff], existing_items: Sequence[RunItem] | None = None, ) -> ProcessedResponse: items: list[RunItem] = [] run_handoffs = [] functions = [] computer_actions = [] custom_tool_calls = [] local_shell_calls = [] shell_calls = [] apply_patch_calls = [] mcp_approval_requests = [] tools_used: list[str] = [] handoff_map = {handoff.tool_name: handoff for handoff in handoffs} function_map = build_function_tool_lookup_map( [tool for tool in all_tools if isinstance(tool, FunctionTool)] ) custom_tool_map = {tool.name: tool for tool in all_tools if isinstance(tool, CustomTool)} computer_tool = next((tool for tool in all_tools if isinstance(tool, ComputerTool)), None) local_shell_tool = next((tool for tool in all_tools if isinstance(tool, LocalShellTool)), None) shell_tool = next((tool for tool in all_tools if isinstance(tool, ShellTool)), None) apply_patch_tool = next((tool for tool in all_tools if isinstance(tool, ApplyPatchTool)), None) hosted_mcp_server_map = { tool.tool_config["server_label"]: tool for tool in all_tools if isinstance(tool, HostedMCPTool) } hosted_mcp_tool_metadata = collect_mcp_list_tools_metadata(existing_items or ()) hosted_mcp_tool_metadata.update(collect_mcp_list_tools_metadata(response.output)) def _dump_output_item(raw_item: Any) -> dict[str, Any]: if isinstance(raw_item, dict): return dict(raw_item) if hasattr(raw_item, "model_dump"): dumped = cast(Any, raw_item).model_dump(exclude_unset=True) if isinstance(dumped, Mapping): return dict(dumped) return {"type": get_mapping_or_attr(raw_item, "type")} return { "type": get_mapping_or_attr(raw_item, "type"), "id": get_mapping_or_attr(raw_item, "id"), } for output in response.output: output_type = get_mapping_or_attr(output, "type") logger.debug( "Processing output item type=%s class=%s", output_type, output.__class__.__name__ if hasattr(output, "__class__") else type(output), ) if output_type == "shell_call": if isinstance(output, dict): shell_call_raw = dict(output) elif hasattr(output, "model_dump"): shell_call_raw = cast(Any, output).model_dump(exclude_unset=True) else: shell_call_raw = { "type": "shell_call", "id": get_mapping_or_attr(output, "id"), "call_id": get_mapping_or_attr(output, "call_id"), "status": get_mapping_or_attr(output, "status"), "action": get_mapping_or_attr(output, "action"), "environment": get_mapping_or_attr(output, "environment"), "created_by": get_mapping_or_attr(output, "created_by"), } shell_call_raw.pop("created_by", None) items.append(ToolCallItem(raw_item=cast(Any, shell_call_raw), agent=agent)) if not shell_tool: tools_used.append("shell") _error_tracing.attach_error_to_current_span( SpanError( message="Shell tool not found", data={}, ) ) raise ModelBehaviorError("Model produced shell call without a shell tool.") tools_used.append(shell_tool.name) shell_environment = shell_tool.environment if shell_environment is None or shell_environment["type"] != "local": logger.debug( "Skipping local shell execution for hosted shell tool %s", shell_tool.name ) continue if shell_tool.executor is None: _error_tracing.attach_error_to_current_span( SpanError( message="Local shell executor not found", data={}, ) ) raise ModelBehaviorError( "Model produced local shell call without a local shell executor." ) call_identifier = get_mapping_or_attr(output, "call_id") logger.debug("Queuing shell_call %s", call_identifier) shell_calls.append(ToolRunShellCall(tool_call=output, shell_tool=shell_tool)) continue if output_type == "shell_call_output" and isinstance( output, dict | ResponseFunctionShellToolCallOutput ): tools_used.append(shell_tool.name if shell_tool else "shell") if isinstance(output, dict): shell_output_raw = dict(output) else: shell_output_raw = output.model_dump(exclude_unset=True) shell_output_raw.pop("created_by", None) shell_outputs = shell_output_raw.get("output") if isinstance(shell_outputs, list): for shell_output in shell_outputs: if isinstance(shell_output, dict): shell_output.pop("created_by", None) items.append( ToolCallOutputItem( raw_item=cast(Any, shell_output_raw), output=shell_output_raw.get("output"), agent=agent, ) ) continue if output_type == "apply_patch_call": if isinstance(output, dict): apply_patch_call_raw = dict(output) elif hasattr(output, "model_dump"): apply_patch_call_raw = cast(Any, output).model_dump(exclude_unset=True) else: apply_patch_call_raw = { "type": "apply_patch_call", "id": get_mapping_or_attr(output, "id"), "call_id": get_mapping_or_attr(output, "call_id"), "status": get_mapping_or_attr(output, "status"), "operation": get_mapping_or_attr(output, "operation"), "created_by": get_mapping_or_attr(output, "created_by"), } apply_patch_call_raw.pop("created_by", None) items.append(ToolCallItem(raw_item=cast(Any, apply_patch_call_raw), agent=agent)) if apply_patch_tool: tools_used.append(apply_patch_tool.name) call_identifier = get_mapping_or_attr(apply_patch_call_raw, "call_id") logger.debug("Queuing apply_patch_call %s", call_identifier) apply_patch_calls.append( ToolRunApplyPatchCall( tool_call=apply_patch_call_raw, apply_patch_tool=apply_patch_tool, ) ) else: tools_used.append("apply_patch") _error_tracing.attach_error_to_current_span( SpanError( message="Apply patch tool not found", data={}, ) ) raise ModelBehaviorError( "Model produced apply_patch call without an apply_patch tool." ) continue if output_type == "compaction": if isinstance(output, dict): compaction_raw = dict(output) elif isinstance(output, ResponseCompactionItem): compaction_raw = output.model_dump(exclude_unset=True) else: logger.warning("Unexpected compaction output type, ignoring: %s", type(output)) continue compaction_raw.pop("created_by", None) items.append( CompactionItem(agent=agent, raw_item=cast(TResponseInputItem, compaction_raw)) ) continue if output_type == "tool_search_call": tool_search_call_raw = coerce_tool_search_call_raw_item(output) if get_mapping_or_attr(tool_search_call_raw, "execution") == "client": raise ModelBehaviorError( "Client-executed tool_search calls are not supported by the standard " "agent runner. Handle the tool_search_call yourself and return a matching " "tool_search_output item with the same call_id." ) items.append(ToolSearchCallItem(raw_item=tool_search_call_raw, agent=agent)) tools_used.append("tool_search") continue if output_type == "tool_search_output": items.append( ToolSearchOutputItem( raw_item=coerce_tool_search_output_raw_item(output), agent=agent, ) ) tools_used.append("tool_search") continue if isinstance(output, ResponseOutputMessage): items.append(MessageOutputItem(raw_item=output, agent=agent)) elif isinstance(output, ResponseFileSearchToolCall): items.append(ToolCallItem(raw_item=output, agent=agent)) tools_used.append("file_search") elif isinstance(output, ResponseFunctionWebSearch): items.append(ToolCallItem(raw_item=output, agent=agent)) tools_used.append("web_search") elif isinstance(output, ResponseReasoningItem): items.append(ReasoningItem(raw_item=output, agent=agent)) elif isinstance(output, ResponseComputerToolCall): items.append(ToolCallItem(raw_item=output, agent=agent)) if not computer_tool: tools_used.append("computer") _error_tracing.attach_error_to_current_span( SpanError( message="Computer tool not found", data={}, ) ) raise ModelBehaviorError("Model produced computer action without a computer tool.") tools_used.append(computer_tool.name) computer_actions.append( ToolRunComputerAction(tool_call=output, computer_tool=computer_tool) ) elif isinstance(output, McpApprovalRequest): items.append(MCPApprovalRequestItem(raw_item=output, agent=agent)) if output.server_label not in hosted_mcp_server_map: _error_tracing.attach_error_to_current_span( SpanError( message="MCP server label not found", data={"server_label": output.server_label}, ) ) raise ModelBehaviorError(f"MCP server label {output.server_label} not found") server = hosted_mcp_server_map[output.server_label] mcp_approval_requests.append( ToolRunMCPApprovalRequest( request_item=output, mcp_tool=server, ) ) if not server.on_approval_request: logger.debug( "Hosted MCP server %s has no on_approval_request hook; approvals will be " "surfaced as interruptions for the caller to handle.", output.server_label, ) elif isinstance(output, McpListTools): items.append(MCPListToolsItem(raw_item=output, agent=agent)) elif isinstance(output, McpCall): metadata = hosted_mcp_tool_metadata.get((output.server_label, output.name)) items.append( ToolCallItem( raw_item=output, agent=agent, description=metadata.description if metadata is not None else None, title=metadata.title if metadata is not None else None, tool_origin=ToolOrigin( type=ToolOriginType.MCP, mcp_server_name=output.server_label, ), ) ) tools_used.append("mcp") elif isinstance(output, ImageGenerationCall): items.append(ToolCallItem(raw_item=output, agent=agent)) tools_used.append("image_generation") elif isinstance(output, ResponseCodeInterpreterToolCall): items.append(ToolCallItem(raw_item=output, agent=agent)) tools_used.append("code_interpreter") elif isinstance(output, LocalShellCall): items.append(ToolCallItem(raw_item=output, agent=agent)) if local_shell_tool: tools_used.append("local_shell") local_shell_calls.append( ToolRunLocalShellCall(tool_call=output, local_shell_tool=local_shell_tool) ) elif shell_tool: tools_used.append(shell_tool.name) shell_calls.append(ToolRunShellCall(tool_call=output, shell_tool=shell_tool)) else: tools_used.append("local_shell") _error_tracing.attach_error_to_current_span( SpanError( message="Local shell tool not found", data={}, ) ) raise ModelBehaviorError( "Model produced local shell call without a local shell tool." ) elif isinstance(output, ResponseCustomToolCall): custom_tool = custom_tool_map.get(output.name) if custom_tool is not None: items.append(ToolCallItem(raw_item=cast(Any, output), agent=agent)) tools_used.append(custom_tool.name) custom_tool_calls.append(ToolRunCustom(tool_call=output, custom_tool=custom_tool)) elif is_apply_patch_name(output.name, apply_patch_tool): parsed_operation = parse_apply_patch_custom_input(output.input) pseudo_call = { "type": "apply_patch_call", "call_id": output.call_id, **parsed_operation, } items.append(ToolCallItem(raw_item=cast(Any, pseudo_call), agent=agent)) if apply_patch_tool: tools_used.append(apply_patch_tool.name) apply_patch_calls.append( ToolRunApplyPatchCall( tool_call=pseudo_call, apply_patch_tool=apply_patch_tool, ) ) else: tools_used.append("apply_patch") _error_tracing.attach_error_to_current_span( SpanError( message="Apply patch tool not found", data={}, ) ) raise ModelBehaviorError( "Model produced apply_patch call without an apply_patch tool." ) else: items.append(ToolCallItem(raw_item=cast(Any, output), agent=agent)) _error_tracing.attach_error_to_current_span( SpanError( message="Custom tool not found", data={"tool_name": output.name}, ) ) raise ModelBehaviorError(f"Tool {output.name} not found in agent {agent.name}") elif ( isinstance(output, ResponseFunctionToolCall) and is_apply_patch_name(output.name, apply_patch_tool) and get_function_tool_lookup_key_for_call(output) not in function_map ): parsed_operation = parse_apply_patch_function_args(output.arguments) pseudo_call = { "type": "apply_patch_call", "call_id": output.call_id, "operation": parsed_operation, } items.append(ToolCallItem(raw_item=cast(Any, pseudo_call), agent=agent)) if apply_patch_tool: tools_used.append(apply_patch_tool.name) apply_patch_calls.append( ToolRunApplyPatchCall(tool_call=pseudo_call, apply_patch_tool=apply_patch_tool) ) else: tools_used.append("apply_patch") _error_tracing.attach_error_to_current_span( SpanError( message="Apply patch tool not found", data={}, ) ) raise ModelBehaviorError( "Model produced apply_patch call without an apply_patch tool." ) continue elif not isinstance(output, ResponseFunctionToolCall): logger.warning("Unexpected output type, ignoring: %s", type(output)) continue if not isinstance(output, ResponseFunctionToolCall): continue tools_used.append(get_tool_call_trace_name(output) or output.name) qualified_output_name = get_tool_call_qualified_name(output) if qualified_output_name == output.name and output.name in handoff_map: items.append(HandoffCallItem(raw_item=output, agent=agent)) handoff = ToolRunHandoff( tool_call=output, handoff=handoff_map[output.name], ) run_handoffs.append(handoff) else: lookup_key = get_function_tool_lookup_key_for_call(output) func_tool = function_map.get(lookup_key) if lookup_key is not None else None if func_tool is None: if output_schema is not None and output.name == "json_tool_call": synthetic_tool = build_litellm_json_tool_call(output) items.append( ToolCallItem( raw_item=output, agent=agent, description=synthetic_tool.description, tool_origin=get_function_tool_origin(synthetic_tool), ) ) functions.append( ToolRunFunction( tool_call=output, function_tool=synthetic_tool, ) ) continue _error_tracing.attach_error_to_current_span( SpanError( message="Tool not found", data={"tool_name": qualified_output_name or output.name}, ) ) error = ( f"Tool {qualified_output_name or output.name} not found in agent {agent.name}" ) raise ModelBehaviorError(error) items.append( ToolCallItem( raw_item=output, agent=agent, description=func_tool.description, title=func_tool._mcp_title, tool_origin=get_function_tool_origin(func_tool), ) ) functions.append( ToolRunFunction( tool_call=output, function_tool=func_tool, ) ) return ProcessedResponse( new_items=items, handoffs=run_handoffs, functions=functions, computer_actions=computer_actions, custom_tool_calls=custom_tool_calls, local_shell_calls=local_shell_calls, shell_calls=shell_calls, apply_patch_calls=apply_patch_calls, tools_used=tools_used, mcp_approval_requests=mcp_approval_requests, interruptions=[], ) async def get_single_step_result_from_response( *, bindings: AgentBindings[TContext], all_tools: list[Tool], original_input: str | list[TResponseInputItem], pre_step_items: list[RunItem], new_response: ModelResponse, output_schema: AgentOutputSchemaBase | None, handoffs: list[Handoff], hooks: RunHooks[TContext], context_wrapper: RunContextWrapper[TContext], run_config: RunConfig, tool_use_tracker, server_manages_conversation: bool = False, event_queue: asyncio.Queue[StreamEvent | QueueCompleteSentinel] | None = None, before_side_effects: Callable[[], Awaitable[None]] | None = None, ) -> SingleStepResult: item_agent = bindings.public_agent processed_response = process_model_response( agent=item_agent, all_tools=all_tools, response=new_response, output_schema=output_schema, handoffs=handoffs, existing_items=pre_step_items, ) if before_side_effects is not None: await before_side_effects() tool_use_tracker.record_processed_response(item_agent, processed_response) if event_queue is not None and processed_response.new_items: handoff_items = [ item for item in processed_response.new_items if isinstance(item, HandoffCallItem) ] if handoff_items: stream_step_items_to_queue(cast(list[RunItem], handoff_items), event_queue) return await execute_tools_and_side_effects( bindings=bindings, original_input=original_input, pre_step_items=pre_step_items, new_response=new_response, processed_response=processed_response, output_schema=output_schema, hooks=hooks, context_wrapper=context_wrapper, run_config=run_config, server_manages_conversation=server_manages_conversation, )