Failed to yield ToolReturnPart streaming with structured output type via tool call

[lorr1]

Initial Checks

• I confirm that I'm using the latest version of Pydantic AI
• I confirm that I searched for my issue in https://github.com/pydantic/pydantic-ai/issues before opening this issue

Description

When building an agent that uses tool calls to handle structured outputs, the final result tool return is never streamed back, it is only returned when the complete list of messages is returned. Specifically this part

pydantic-ai/pydantic_ai_slim/pydantic_ai/_agent_graph.py

Line 636 in f5f7b80

part = _messages.ToolReturnPart(

. The tool return is created and added to the outputs but not yielded.

This may have been on purpose, and I'd like to understand why. In most use cases, structured outputs are one-shot runs where there is no chat back with the model. So not including the tool return is fine. But I wanted to chat with the model and assumed the streamed responses would give me an equivalent history for chat applications as calling all_messages().

Without yielding the tool return, the streamed responses cause a dangling tool call which causes issues if you try to feed that back as message history.

Happy to submit a PR to add the yielding if you folks want. Let me know what you think!

Example Code

# Run in a notebook, hence nest_asyncio
# Modified from https://ai.pydantic.dev/agents/#streaming-all-events-and-output
import asyncio
from pydantic import BaseModel
from pydantic_ai import Agent, StructuredDict
from pydantic_ai.messages import FinalResultEvent, FunctionToolCallEvent, FunctionToolResultEvent
import nest_asyncio

nest_asyncio.apply()

async def run_agent(agent, prompt):
    output_messages = []
    # Begin a node-by-node, streaming iteration
    async with agent.iter(prompt) as run:
        async for node in run:
            if Agent.is_user_prompt_node(node):
                # A user prompt node => The user has provided input
                output_messages.append(f'=== UserPromptNode: {node.user_prompt} ===')
            elif Agent.is_model_request_node(node):
                # A model request node => We can stream tokens from the model's request
                async with node.stream(run.ctx) as request_stream:
                    async for event in request_stream:
                        pass
                    output_messages.append(request_stream.get())
            elif Agent.is_call_tools_node(node):
                # A handle-response node => The model returned some data, potentially calls a tool
                async with node.stream(run.ctx) as handle_stream:
                    async for event in handle_stream:
                        if isinstance(event, FunctionToolCallEvent):
                            output_messages.append(event)
                        elif isinstance(event, FunctionToolResultEvent):
                            output_messages.append(event.result)
            elif Agent.is_end_node(node):
                # Once an End node is reached, the agent run is complete
                assert run.result is not None
                assert run.result.output == node.data.output
                output_messages.append(run.result.output)
    return output_messages


class TestSchema(BaseModel):
    user_name: str
    user_description: str


json_schema = TestSchema.model_json_schema()
agent = Agent(
    "openai:gpt-4o",
    system_prompt="Either ask for clarification or respond with the final result tool.",
    output_type=StructuredDict(json_schema),
)
# Full run sync to get all the messages
result = agent.run_sync("Hi")

for msg in result.all_messages():
    # The first message with be ModelRequest for system promptand user prompt
    # Second is the tool call
    # Third is the tool return
    print(msg)
    print()

print("==============")
msgs = asyncio.run(run_agent(agent, "Hi"))
for msg in msgs:
    # First is user prompt
    # Second is tool call
    # Third is just output - no tool return
    print(msg)
    print()

"""RESULTS
ModelRequest(parts=[SystemPromptPart(content='Either ask for clarification or respond with the final result tool.', timestamp=datetime.datetime(2025, 8, 22, 5, 28, 23, 285240, tzinfo=datetime.timezone.utc)), UserPromptPart(content='Hi', timestamp=datetime.datetime(2025, 8, 22, 5, 28, 23, 285246, tzinfo=datetime.timezone.utc))])

ModelResponse(parts=[ToolCallPart(tool_name='final_result', args='{"user_name":"Hello, how can I assist you today?","user_description":"User greeted."}', tool_call_id='call_xxFXtp52ubL9XPdC0XK21EbD')], usage=RequestUsage(input_tokens=64, output_tokens=30, 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, 8, 22, 5, 28, 23, tzinfo=TzInfo(UTC)), provider_request_id='chatcmpl-C7Ec3CbnlDC0oZGYKS7xxQQdFtaCB')

ModelRequest(parts=[ToolReturnPart(tool_name='final_result', content='Final result processed.', tool_call_id='call_xxFXtp52ubL9XPdC0XK21EbD', timestamp=datetime.datetime(2025, 8, 22, 5, 28, 24, 19144, tzinfo=datetime.timezone.utc))])

==============
=== UserPromptNode: Hi ===

ModelResponse(parts=[ToolCallPart(tool_name='final_result', args='', tool_call_id='call_Dt32fw0NnEoOKhPLdtpSeFsi')], usage=RequestUsage(details={'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}), model_name='gpt-4o', timestamp=datetime.datetime(2025, 8, 22, 5, 28, 24, tzinfo=TzInfo(UTC)))

{'user_name': 'Hi', 'user_description': 'An initial greeting message.'}
"""
# I am expecting a toolreturnpart object in the second response just like in the first

Python, Pydantic AI & LLM client version

Python version: 3.13.5 (main, Jul 11 2025, 22:26:07) [Clang 20.1.4 ]
Pydantic AI version: 0.7.4
LLM Client OpenAI version: 1.100.2

[DouweM]

@lorr1 Output tools currently only generate a FunctionToolCallEvent/FunctionToolResultEvent pair when output validation failed, because in that case we know there'll definitely be a subsequent request to the model, and the event-based message history would be expected to be complete without missing tool call results in the middle.

Outside of that scenario, the reasoning to omit the events presumably has been that an output tool call is not "really" a tool call and typically wouldn't be shown to the user as one -- they'd just see the final result itself.

I definitely can see the argument that the event-based message history should be equivalent to that in all_messages(), and that it'd be up to the event consumer to filter out the output tool calls/results if they don't want to show them. Changing that now would constitute a breaking change, though, as people's code wouldn't be expecting these calls to tools they didn't explicitly define themselves (and weren't expecting to show to the user). Of course, user code already has to account for output tools in the PartStartEvent(ToolCallPart)s that are streamed during the model response, so this wouldn't necessarily be a huge deal.

What would you think of keeping the behavior as is, but making it explicit that for message history purposes, you should always use all_messages() instead of reconstructing it manually from events? One possible issue with that would be if your frontend creates the subsequent agent run message history and only gets to see the event stream, never the full response/request objects (as with our AG-UI integration).

If we decide to change this, we'll want to do it this week before v1, after which the bar to introduce breaking changes will become higher.

[lorr1]

I guess I wanted to dig more into one thing you said around always using the all_messages() method always for history. So I dug into refactoring my code to use all_messages() for the history rather than have the history mimic exactly what's streamed to the UI. The individual parts match (except for the tool return bit from the final_result tool) but the grouping is different. In all_messages(), for example, all tool returns (if the model makes multiple tool calls in parallel) are all grouped into one ModelRequest. When streaming to the UI, each tool return is streamed as it's finished in its own ModelRequest. We want to keep the UI fed with data as soon as it comes in. As far as I can tell from looking at map_messages, this grouping difference makes no difference in how you call the model. But, it could change behavior in the UI. Or rather, it would require the UI to handle two different message groupings depending on if the result was streamed versus pulled from history.

I'm mainly trying to point out that the contract between the all_messages() and the event stream is already kind of broken independent of the structured output tool call stuff.

From a user's, I actually would have been supportive if you guys decided to swallow the tool call and return and turn it into a final model message or something (which would make it feel more like the other methods for dealing with structured outputs). It is this weird final tool call which most people probably don't want to deal with or understand. And then at least the event stream would result in a consistent state with no dangling tool call. From a model perspective, you don't want to do that as it could cause quality issues in a longer chat if the model never saw itself making the final tool call.

Anyways, I'm still in support of adding it as a yield and can have a PR out tomorrow (or maybe your today depending on your timezone). Also get if you decide to not fix. It's an edge case I've already worked around.

[DouweM]

In all_messages(), for example, all tool returns (if the model makes multiple tool calls in parallel) are all grouped into one ModelRequest. When streaming to the UI, each tool return is streamed as it's finished in its own ModelRequest.

@lorr1 I'm not sure I follow -- when streaming events, we don't emit "ModelRequest boundaries" just tool results, so how do you mean each tool return is "in its own ModelRequest"? The message grouping would be up to your code, right?

[lorr1]

Ya I didn't do a great job explaining. So in the event stream, you yield FunctionToolResultEvent which contains ToolReturnPart result (assuming no model retry). Each tool return gets its own FunctionToolResultEvent. As soon as we get this ToolReturnPart, we yield it to the FE wrapped in a ModelRequest for type consistency so everything is a ModelMessage. This is a choice in our code. So from the FEs perspective, it gets a list of ModelRequest. This history formats and groups the tool returns differently. It's up to the pydantic ai users to ensure the event stream matches the history because if you just yield each event as it comes in, it won't be the same.

[DouweM]

we yield it to the FE wrapped in a ModelRequest for type consistency so everything is a ModelMessage

@lorr1 Hmm, then how do you handle model response events (text, thinking, and tool call parts) streamed to the frontend? Are those individually wrapped in ModelResponse objects as well? Or is the response not currently streamed but sent in full?

[lorr1]

For model responses (thinking, tool calls, etc), we use

agent_stream.stream_responses()

And yield each msg from that stream as it is.

I think this streams each new token from the model but gathers the response in one final model message. I think? But it's done automatically in the stream_responses() method.

We used what streaming methods I could find from you guys to build the stream. As soon as anything comes back, we yield pretty much.

[DouweM]

@lorr1 Ah yeah, that builds a ModelResponse for everything that's been received up to this point, so cumulatively rather than using deltas like you'd get with https://ai.pydantic.dev/agents/#streaming-all-events. If your frontend code is already prepared to treat new ModelResponse/ModelRequest messages as replacing the most recent one, you could do what you're doing in building ModelRequest for each FunctionToolCallEvent/FunctionToolResultEvent, but then follow it up with result.all_messages()[-1] once the run is complete so that the frontend also gets the complete one that will have the output tool message in it.