Parallel output tool calls with `end_strategy='exhaustive'` should call all output functions

[Danipulok]

Question

Hey here!

I was testing how Pydantic AI tools calls are done, and noticed that when LLM responses with 2 tool calls (both of which are output type), only one tool is called. Is it intended or is there a way to configure it?

I understand this is likely intended, but seems a little bit non-intuitive IMHO (but I see the logic)

Additional Context

MRE

import asyncio
import os

from pydantic import BaseModel
from pydantic_ai import Agent, ModelSettings, ToolOutput
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider


class TextMessage(BaseModel):
    text: str | None = None


async def send_text_message(
    message: TextMessage,
) -> None:
    """Send a text message."""
    print(f"\nText Message: {message}")


class QuickRepliesMessage(BaseModel):
    text: str | None = None
    quick_replies: list[str] | None = None


async def send_quick_replies_message(
    message: QuickRepliesMessage,
) -> None:
    """Send a quick replies message."""
    print(f"\nQuick Replies Message: {message}")


async def main() -> None:
    api_key = os.environ["OPENAI_API_KEY"]
    model = OpenAIChatModel(
        "gpt-4o",
        provider=OpenAIProvider(api_key=api_key),
        settings=ModelSettings(
            temperature=0.1,
        ),
    )
    output_type = [
        ToolOutput(send_text_message, name="send_text_message"),
        ToolOutput(send_quick_replies_message, name="send_quick_replies_message"),
    ]
    agent = Agent(
        model,
        output_type=output_type,
        instructions="For response, call both `send_text_message` and `send_quick_replies_message` tools in parallel",
    )

    user_prompt = "Tell me about Python"

    async with agent.run_stream(user_prompt) as run:
        async for output in run.stream_responses():
            model_response, is_last_message = output
            print(model_response, is_last_message, end="\n\n")


if __name__ == "__main__":
    asyncio.run(main())

Console:

ModelResponse(parts=[ToolCallPart(tool_name='send_text_message', args='{"text": "Python is a high-level, interpreted programming language known for its easy-to-read syntax and broad applicability. Created in the late 1980s by Guido van Rossum and released in 1991, Python emphasizes code readability and simplicity. It supports various programming paradigms, including procedural, object-oriented, and functional programming."}', tool_call_id='call_RcMJNbdMVPfB3jedSL8j4ioE'), ToolCallPart(tool_name='send_quick_replies_message', args='{"text": "Would you like to know about something specific in Python?", "quick_replies": ["Python Syntax", "Libraries", "Use Cases", "History", "Installation"]}', tool_call_id='call_xVHAJyJ6HDq8QzBM0cr7VC6j')], usage=RequestUsage(input_tokens=110, output_tokens=139, details={'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}), model_name='gpt-4o-2024-08-06', timestamp=datetime.datetime(2025, 11, 20, 4, 36, 45, tzinfo=TzInfo(0)), provider_name='openai', provider_details={'finish_reason': 'tool_calls'}, provider_response_id='chatcmpl-CdqhRR4PuMcDZLezcHUd2x00iOSh9', finish_reason='tool_call') True

Text Message: text='Python is a high-level, interpreted programming language known for its easy-to-read syntax and broad applicability. Created in the late 1980s by Guido van Rossum and released in 1991, Python emphasizes code readability and simplicity. It supports various programming paradigms, including procedural, object-oriented, and functional programming.'

Process finished with exit code 0

[DouweM]

@Danipulok Check out Agent(end_strategy=...):

• https://ai.pydantic.dev/api/agent/#pydantic_ai.agent.Agent.end_strategy
• https://ai.pydantic.dev/api/agent/#pydantic_ai.agent.EndStrategy

[DouweM]

@Danipulok There's a good chance the exhaustive mode will only ensure all function tools are executed, not all output functions. If you can confirm that's the case, we can use this issue to make sure we call those as well.

[Danipulok]

@DouweM thank you very much for response!
As you suspected, it doesn't seem to work

Here's the changed code:

    agent = Agent(
        model,
        output_type=output_type,
        instructions="For response, call both `send_text_message` and `send_quick_replies_message` tools in parallel",
        end_strategy="exhaustive",  # comment to use the default "early" strategy
    )

Here's the output:

ModelResponse(parts=[ToolCallPart(tool_name='send_text_message', args='{"text": "Python is a high-level, interpreted programming language known for its readability and simplicity. It supports multiple programming paradigms, including procedural, object-oriented, and functional programming. Python is widely used in web development, data analysis, artificial intelligence, scientific computing, and more."}', tool_call_id='call_iqyle7PEgwRrKPIB9uhMaEQ9'), ToolCallPart(tool_name='send_quick_replies_message', args='{"text": "Would you like to know more about Python?", "quick_replies": ["History of Python", "Python Features", "Python Applications", "Learning Resources"]}', tool_call_id='call_50K33vhEJ7lyfk6cjulek3hZ')], usage=RequestUsage(input_tokens=124, output_tokens=127, details={'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}), model_name='gpt-4o-2024-08-06', timestamp=datetime.datetime(2025, 11, 20, 16, 42, 7, tzinfo=TzInfo(0)), provider_name='openai', provider_details={'finish_reason': 'tool_calls'}, provider_response_id='chatcmpl-Ce21PzXA1xj1yLIcXxGBJyZR5KrEy', finish_reason='tool_call') True

Text Message: text='Python is a high-level, interpreted programming language known for its readability and simplicity. It supports multiple programming paradigms, including procedural, object-oriented, and functional programming. Python is widely used in web development, data analysis, artificial intelligence, scientific computing, and more.'

Process finished with exit code 0

It works the same as end_strategy="early" for me
I've updated the example above and added temperature to make it more reproducible

If I use this prompt:

    agent = Agent(
        model,
        output_type=output_type,
        instructions="For response, call at first `send_quick_replies_message` and then `send_text_message`, both in parallel",
        end_strategy="exhaustive",  # comment to use the default "early" strategy
    )

Then the send_quick_replies_message is called and not send_text_message. So it works the same as end_strategy="early" and calls only the first output type tool

[DouweM]

@Danipulok Thanks for confirming. The relevant logic is here, first for handling output tool calls, and then function tools (which do handle end_strategy):

pydantic-ai/pydantic_ai_slim/pydantic_ai/_agent_graph.py

Lines 816 to 871 in d578b42

	# First, we handle output tool calls
	for call in tool_calls_by_kind['output']:
	if final_result:
	if final_result.tool_call_id == call.tool_call_id:
	part = _messages.ToolReturnPart(
	tool_name=call.tool_name,
	content='Final result processed.',
	tool_call_id=call.tool_call_id,
	)
	else:
	yield _messages.FunctionToolCallEvent(call)
	part = _messages.ToolReturnPart(
	tool_name=call.tool_name,
	content='Output tool not used - a final result was already processed.',
	tool_call_id=call.tool_call_id,
	)
	yield _messages.FunctionToolResultEvent(part)
	output_parts.append(part)
	else:
	try:
	result_data = await tool_manager.handle_call(call)
	except exceptions.UnexpectedModelBehavior as e:
	ctx.state.increment_retries(
	ctx.deps.max_result_retries, error=e, model_settings=ctx.deps.model_settings
	)
	raise e # pragma: lax no cover
	except ToolRetryError as e:
	ctx.state.increment_retries(
	ctx.deps.max_result_retries, error=e, model_settings=ctx.deps.model_settings
	)
	yield _messages.FunctionToolCallEvent(call)
	output_parts.append(e.tool_retry)
	yield _messages.FunctionToolResultEvent(e.tool_retry)
	else:
	part = _messages.ToolReturnPart(
	tool_name=call.tool_name,
	content='Final result processed.',
	tool_call_id=call.tool_call_id,
	)
	output_parts.append(part)
	final_result = result.FinalResult(result_data, call.tool_name, call.tool_call_id)
	# Then, we handle function tool calls
	calls_to_run: list[_messages.ToolCallPart] = []
	if final_result and ctx.deps.end_strategy == 'early':
	for call in tool_calls_by_kind['function']:
	output_parts.append(
	_messages.ToolReturnPart(
	tool_name=call.tool_name,
	content='Tool not executed - a final result was already processed.',
	tool_call_id=call.tool_call_id,
	)
	)
	else:
	calls_to_run.extend(tool_calls_by_kind['function'])

The goal would be to update the output tool logic to also call tool_manager.handle_call if there's already a final_result but end_strategy == 'exhaustive'

Would you be up for submitting an (AI-assisted) PR?

[Danipulok]

@DouweM, I guess I will at least try :D

It would be interesting and I would be glad to contribute to pydantic-ai

I will mention this issue and you when I will create a PR, thank you!