Function invoke_function_tool

src/agents/tool.py:1806–1847

Invoke a function tool, enforcing timeout configuration when provided.

(
    *,
    function_tool: FunctionTool,
    context: ToolContext[Any],
    arguments: str,
)

async def invoke_function_tool(
    *,
    function_tool: FunctionTool,
    context: ToolContext[Any],
    arguments: str,
) -> Any:
    """Invoke a function tool, enforcing timeout configuration when provided."""
    invoke_context = _get_function_tool_invoke_context(function_tool, context)
    timeout_seconds = function_tool.timeout_seconds
    if timeout_seconds is None:
        return await function_tool.on_invoke_tool(cast(Any, invoke_context), arguments)

    tool_task: asyncio.Future[Any] = asyncio.ensure_future(
        function_tool.on_invoke_tool(cast(Any, invoke_context), arguments)
    )
    try:
        return await asyncio.wait_for(tool_task, timeout=timeout_seconds)
    except asyncio.TimeoutError as exc:
        if tool_task.done() and not tool_task.cancelled():
            tool_exception = tool_task.exception()
            if tool_exception is None:
                return tool_task.result()
            raise tool_exception from None

        timeout_error = ToolTimeoutError(
            tool_name=function_tool.name,
            timeout_seconds=timeout_seconds,
        )
        if function_tool.timeout_behavior == "raise_exception":
            raise timeout_error from exc

        timeout_error_function = function_tool.timeout_error_function
        if timeout_error_function is None:
            return default_tool_timeout_error_message(
                tool_name=function_tool.name,
                timeout_seconds=timeout_seconds,
            )

        timeout_result = timeout_error_function(context, timeout_error)
        if inspect.isawaitable(timeout_result):
            return await timeout_result
        return timeout_result


@overload