Using Function Model with streaming responses

[tenzo]

Question

Hello, could anyone provide an example of using the FunctionModel for testing tools being called? I'm having various issues as I call it like:

with my_agent.override(
        model=FunctionModel(stream_function=call_my_tool)
):

And the call_my_tool should be an async iterator yielding one of 3 types:

StreamFunctionDef: TypeAlias = Callable[
    [list[ModelMessage], AgentInfo], AsyncIterator[Union[str, DeltaToolCalls, DeltaThinkingCalls]]
]

so I assume I should yield DeltaToolCalls (=dict[int, DeltaToolCall] but if my function returns something like:

        yield {
            1: DeltaToolCall(
                name='my_tool',
            )
        }

I'm having errors with iterator being exhausted:

ValueError: Stream function must return at least one item

Can anyone provide a working example with FunctionModel and iter in a test?
AFAIK there's no such example available.

Additional Context

PydanticAI 0.6.2
Python 3.13

[DouweM]

@tenzo You can see an example of this in our test suite:

pydantic-ai/tests/test_streaming.py

Lines 1115 to 1148 in d6d7f37

	async def test_iter_stream_output_tool_dont_hit_retry_limit():
	class CityLocation(BaseModel):
	city: str
	country: str | None = None
	async def text_stream(_messages: list[ModelMessage], agent_info: AgentInfo) -> AsyncIterator[DeltaToolCalls]:
	"""Stream partial JSON data that will initially fail validation."""
	assert agent_info.output_tools is not None
	assert len(agent_info.output_tools) == 1
	name = agent_info.output_tools[0].name
	yield {0: DeltaToolCall(name=name)}
	yield {0: DeltaToolCall(json_args='{"c')}
	yield {0: DeltaToolCall(json_args='ity":')}
	yield {0: DeltaToolCall(json_args=' "Mex')}
	yield {0: DeltaToolCall(json_args='ico City",')}
	yield {0: DeltaToolCall(json_args=' "cou')}
	yield {0: DeltaToolCall(json_args='ntry": "Mexico"}')}
	agent = Agent(FunctionModel(stream_function=text_stream), output_type=CityLocation)
	async with agent.iter('Generate city info') as run:
	async for node in run:
	if agent.is_model_request_node(node):
	async with node.stream(run.ctx) as stream:
	assert [c async for c in stream.stream_output(debounce_by=None)] == snapshot(
	[
	CityLocation(city='Mex'),
	CityLocation(city='Mexico City'),
	CityLocation(city='Mexico City'),
	CityLocation(city='Mexico City', country='Mexico'),
	CityLocation(city='Mexico City', country='Mexico'),
	]
	)

I'll close this issue assuming that gives you what you need to make it work, but if not please share code that reproduces the issue and I'll reopen.