Function _build_wrapped_function_tool

src/agents/tool.py:599–651

Create a FunctionTool with copied-tool-aware failure handling bound in one place.

(
    *,
    name: str,
    description: str,
    params_json_schema: dict[str, Any],
    invoke_tool_impl: Callable[[ToolContext[Any], str], Awaitable[Any]],
    on_handled_error: Callable[[FunctionTool, Exception, str, ToolContext[Any]], None],
    failure_error_function: ToolErrorFunction | None | object = _UNSET_FAILURE_ERROR_FUNCTION,
    strict_json_schema: bool = True,
    is_enabled: bool | Callable[[RunContextWrapper[Any], AgentBase], MaybeAwaitable[bool]] = True,
    tool_input_guardrails: list[ToolInputGuardrail[Any]] | None = None,
    tool_output_guardrails: list[ToolOutputGuardrail[Any]] | None = None,
    needs_approval: (
        bool | Callable[[RunContextWrapper[Any], dict[str, Any], str], Awaitable[bool]]
    ) = False,
    timeout_seconds: float | None = None,
    timeout_behavior: ToolTimeoutBehavior = "error_as_result",
    timeout_error_function: ToolErrorFunction | None = None,
    defer_loading: bool = False,
    custom_data_extractor: FunctionToolCustomDataExtractor | None = None,
    sync_invoker: bool = False,
    mcp_title: str | None = None,
    tool_origin: ToolOrigin | None = None,
)

def _build_wrapped_function_tool(
    *,
    name: str,
    description: str,
    params_json_schema: dict[str, Any],
    invoke_tool_impl: Callable[[ToolContext[Any], str], Awaitable[Any]],
    on_handled_error: Callable[[FunctionTool, Exception, str, ToolContext[Any]], None],
    failure_error_function: ToolErrorFunction | None | object = _UNSET_FAILURE_ERROR_FUNCTION,
    strict_json_schema: bool = True,
    is_enabled: bool | Callable[[RunContextWrapper[Any], AgentBase], MaybeAwaitable[bool]] = True,
    tool_input_guardrails: list[ToolInputGuardrail[Any]] | None = None,
    tool_output_guardrails: list[ToolOutputGuardrail[Any]] | None = None,
    needs_approval: (
        bool | Callable[[RunContextWrapper[Any], dict[str, Any], str], Awaitable[bool]]
    ) = False,
    timeout_seconds: float | None = None,
    timeout_behavior: ToolTimeoutBehavior = "error_as_result",
    timeout_error_function: ToolErrorFunction | None = None,
    defer_loading: bool = False,
    custom_data_extractor: FunctionToolCustomDataExtractor | None = None,
    sync_invoker: bool = False,
    mcp_title: str | None = None,
    tool_origin: ToolOrigin | None = None,
) -> FunctionTool:
    """Create a FunctionTool with copied-tool-aware failure handling bound in one place."""
    on_invoke_tool = _with_context_function_tool_failure_error_handler(
        invoke_tool_impl,
        on_handled_error,
    )
    if sync_invoker:
        setattr(on_invoke_tool, _SYNC_FUNCTION_TOOL_MARKER, True)

    return set_function_tool_failure_error_function(
        FunctionTool(
            name=name,
            description=description,
            params_json_schema=params_json_schema,
            on_invoke_tool=on_invoke_tool,
            strict_json_schema=strict_json_schema,
            is_enabled=is_enabled,
            tool_input_guardrails=tool_input_guardrails,
            tool_output_guardrails=tool_output_guardrails,
            needs_approval=needs_approval,
            timeout_seconds=timeout_seconds,
            timeout_behavior=timeout_behavior,
            timeout_error_function=timeout_error_function,
            defer_loading=defer_loading,
            custom_data_extractor=custom_data_extractor,
            _mcp_title=mcp_title,
            _tool_origin=tool_origin,
        ),
        failure_error_function,
    )


def get_function_tool_origin(function_tool: FunctionTool) -> ToolOrigin | None: