Repository: pydantic/pydantic-ai
File: pydantic_ai_slim/pydantic_ai/_agent_graph.py (Python)

Code before:
async def _call_tools(
    tool_manager: ToolManager[DepsT],
    tool_calls: list[_messages.ToolCallPart],
    deferred_tool_results: dict[str, DeferredToolResult],
    tracer: Tracer,
    usage_limits: _usage.UsageLimits | None,
    output_parts: list[_messages.ModelRequestPart],
    output_deferred_calls: dict[Literal['external', 'unapproved'], list[_messages.ToolCallPart]],
) -> AsyncIterator[_messages.HandleResponseEvent]:
    tool_parts_by_index: dict[int, _messages.ModelRequestPart] = {}
    user_parts_by_index: dict[int, _messages.UserPromptPart] = {}
    deferred_calls_by_index: dict[int, Literal['external', 'unapproved']] = {}

    for call in tool_calls:
        yield _messages.FunctionToolCallEvent(call)

    with tracer.start_as_current_span(
        'running tools',
        attributes={
            'tools': [call.tool_name for call in tool_calls],
            'logfire.msg': f'running {len(tool_calls)} tool{"" if len(tool_calls) == 1 else "s"}',
        },
    ):

        async def handle_call_or_result(
            coro_or_task: Coroutine[
                Any, Any, tuple[_messages.ToolReturnPart | _messages.RetryPromptPart, _messages.UserPromptPart | None]
            ]
            | Task[tuple[_messages.ToolReturnPart | _messages.RetryPromptPart, _messages.UserPromptPart | None]],
            index: int,
        ) -> _messages.HandleResponseEvent | None:
            try:
                tool_part, tool_user_part = (
                    (await coro_or_task) if isinstance(coro_or_task, Coroutine) else coro_or_task.result()
                )
            except exceptions.CallDeferred:
                deferred_calls_by_index[index] = 'external'
            except exceptions.ApprovalRequired:
                deferred_calls_by_index[index] = 'unapproved'
            else:
                tool_parts_by_index[index] = tool_part
                if tool_user_part:
                    user_parts_by_index[index] = tool_user_part

                return _messages.FunctionToolResultEvent(tool_part)

        if tool_manager.should_sequential_tool_call(tool_calls):
            for call in tool_calls:
                index = tool_calls.index(call)
                if event := await handle_call_or_result(
                    _call_tool(tool_manager, call, deferred_tool_results.get(call.tool_call_id), usage_limits),

Review comment (suggestion):
Judging by the rest of the codebase, we prefer `Callable[..., Awaitable[...]]` over `Coroutine[...]`

Code after:

async def _call_tools(
    tool_manager: ToolManager[DepsT],
    tool_calls: list[_messages.ToolCallPart],
    deferred_tool_results: dict[str, DeferredToolResult],
    tracer: Tracer,
    usage_limits: _usage.UsageLimits | None,
    output_parts: list[_messages.ModelRequestPart],
    output_deferred_calls: dict[Literal['external', 'unapproved'], list[_messages.ToolCallPart]],
) -> AsyncIterator[_messages.HandleResponseEvent]:
    tool_parts_by_index: dict[int, _messages.ModelRequestPart] = {}
    user_parts_by_index: dict[int, _messages.UserPromptPart] = {}
    deferred_calls_by_index: dict[int, Literal['external', 'unapproved']] = {}

    for call in tool_calls:
        yield _messages.FunctionToolCallEvent(call)

    with tracer.start_as_current_span(
        'running tools',
        attributes={
            'tools': [call.tool_name for call in tool_calls],
            'logfire.msg': f'running {len(tool_calls)} tool{"" if len(tool_calls) == 1 else "s"}',
        },
    ):

        async def handle_call_or_result(
            coro_or_task: Awaitable[
                tuple[_messages.ToolReturnPart | _messages.RetryPromptPart, _messages.UserPromptPart | None]
            ]
            | Task[tuple[_messages.ToolReturnPart | _messages.RetryPromptPart, _messages.UserPromptPart | None]],
            index: int,
        ) -> _messages.HandleResponseEvent | None:
            try:
                tool_part, tool_user_part = (
                    (await coro_or_task) if inspect.isawaitable(coro_or_task) else coro_or_task.result()
                )
            except exceptions.CallDeferred:
                deferred_calls_by_index[index] = 'external'
            except exceptions.ApprovalRequired:
                deferred_calls_by_index[index] = 'unapproved'
            else:
                tool_parts_by_index[index] = tool_part
                if tool_user_part:
                    user_parts_by_index[index] = tool_user_part

                return _messages.FunctionToolResultEvent(tool_part)

        if tool_manager.should_call_sequentially(tool_calls):
            for index, call in enumerate(tool_calls):
                if event := await handle_call_or_result(
                    _call_tool(tool_manager, call, deferred_tool_results.get(call.tool_call_id), usage_limits),