Yield events for unknown tool calls (#1960)

Co-authored-by: Douwe Maan <me@douwe.me>

Author: proever

pydantic_ai_slim/pydantic_ai/_agent_graph.py

@@ -648,6 +648,7 @@ async def process_function_tools(  # noqa C901
             # if tool_name is in output_schema, it means we found a output tool but an error occurred in
             # validation, we don't add another part here
             if output_tool_name is not None:
+                yield _messages.FunctionToolCallEvent(call)
                 if found_used_output_tool:
                     content = 'Output tool not used - a final result was already processed.'
                 else:
@@ -658,9 +659,14 @@ async def process_function_tools(  # noqa C901
                     content=content,
                     tool_call_id=call.tool_call_id,
                 )
+                yield _messages.FunctionToolResultEvent(part, tool_call_id=call.tool_call_id)
                 output_parts.append(part)
         else:
-            output_parts.append(_unknown_tool(call.tool_name, call.tool_call_id, ctx))
+            yield _messages.FunctionToolCallEvent(call)
+
+            part = _unknown_tool(call.tool_name, call.tool_call_id, ctx)
+            yield _messages.FunctionToolResultEvent(part, tool_call_id=call.tool_call_id)
+            output_parts.append(part)
 
     if not calls_to_run:
         return

pydantic_ai_slim/pydantic_ai/messages.py

@@ -1,7 +1,6 @@
 from __future__ import annotations as _annotations
 
 import base64
-import uuid
 from abc import ABC, abstractmethod
 from collections.abc import Sequence
 from dataclasses import dataclass, field, replace
@@ -888,13 +887,13 @@ class FunctionToolCallEvent:
 
     part: ToolCallPart
     """The (function) tool call to make."""
-    call_id: str = field(init=False)
-    """An ID used for matching details about the call to its result. If present, defaults to the part's tool_call_id."""
     event_kind: Literal['function_tool_call'] = 'function_tool_call'
     """Event type identifier, used as a discriminator."""
 
-    def __post_init__(self):
-        self.call_id = self.part.tool_call_id or str(uuid.uuid4())
+    @property
+    def call_id(self) -> str:
+        """An ID used for matching details about the call to its result. If present, defaults to the part's tool_call_id."""
+        return self.part.tool_call_id
 
     __repr__ = _utils.dataclasses_no_defaults_repr
 

tests/test_streaming.py

@@ -6,7 +6,7 @@
 from collections.abc import AsyncIterator
 from copy import deepcopy
 from datetime import timezone
-from typing import Union
+from typing import Any, Union
 
 import pytest
 from inline_snapshot import snapshot
@@ -15,6 +15,8 @@
 from pydantic_ai import Agent, UnexpectedModelBehavior, UserError, capture_run_messages
 from pydantic_ai.agent import AgentRun
 from pydantic_ai.messages import (
+    FunctionToolCallEvent,
+    FunctionToolResultEvent,
     ModelMessage,
     ModelRequest,
     ModelResponse,
@@ -921,3 +923,120 @@ def output_validator(data: OutputType | NotOutputType) -> OutputType | NotOutput
                     async for output in stream.stream_output(debounce_by=None):
                         outputs.append(output)
     assert outputs == [OutputType(value='a (validated)'), OutputType(value='a (validated)')]
+
+
+async def test_unknown_tool_call_events():
+    """Test that unknown tool calls emit both FunctionToolCallEvent and FunctionToolResultEvent during streaming."""
+
+    def call_mixed_tools(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+        """Mock function that calls both known and unknown tools."""
+        return ModelResponse(
+            parts=[
+                ToolCallPart('unknown_tool', {'arg': 'value'}),
+                ToolCallPart('known_tool', {'x': 5}),
+            ]
+        )
+
+    agent = Agent(FunctionModel(call_mixed_tools))
+
+    @agent.tool_plain
+    def known_tool(x: int) -> int:
+        return x * 2
+
+    event_parts: list[Any] = []
+
+    try:
+        async with agent.iter('test') as agent_run:
+            async for node in agent_run:  # pragma: no branch
+                if Agent.is_call_tools_node(node):
+                    async with node.stream(agent_run.ctx) as event_stream:
+                        async for event in event_stream:
+                            event_parts.append(event)
+
+    except UnexpectedModelBehavior:
+        pass
+
+    assert event_parts == snapshot(
+        [
+            FunctionToolCallEvent(
+                part=ToolCallPart(
+                    tool_name='unknown_tool',
+                    args={'arg': 'value'},
+                    tool_call_id=IsStr(),
+                ),
+            ),
+            FunctionToolResultEvent(
+                result=RetryPromptPart(
+                    content="Unknown tool name: 'unknown_tool'. Available tools: known_tool",
+                    tool_name='unknown_tool',
+                    tool_call_id=IsStr(),
+                    timestamp=IsNow(tz=timezone.utc),
+                ),
+                tool_call_id=IsStr(),
+            ),
+            FunctionToolCallEvent(
+                part=ToolCallPart(tool_name='known_tool', args={'x': 5}, tool_call_id=IsStr()),
+            ),
+            FunctionToolResultEvent(
+                result=ToolReturnPart(
+                    tool_name='known_tool',
+                    content=10,
+                    tool_call_id=IsStr(),
+                    timestamp=IsNow(tz=timezone.utc),
+                ),
+                tool_call_id=IsStr(),
+            ),
+            FunctionToolCallEvent(
+                part=ToolCallPart(
+                    tool_name='unknown_tool',
+                    args={'arg': 'value'},
+                    tool_call_id=IsStr(),
+                ),
+            ),
+        ]
+    )
+
+
+async def test_output_tool_validation_failure_events():
+    """Test that output tools that fail validation emit events during streaming."""
+
+    def call_final_result_with_bad_data(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+        """Mock function that calls final_result tool with invalid data."""
+        assert info.output_tools is not None
+        return ModelResponse(
+            parts=[
+                ToolCallPart('final_result', {'bad_value': 'invalid'}),  # Invalid field name
+                ToolCallPart('final_result', {'value': 'valid'}),  # Valid field name
+            ]
+        )
+
+    agent = Agent(FunctionModel(call_final_result_with_bad_data), output_type=OutputType)
+
+    event_parts: list[Any] = []
+    async with agent.iter('test') as agent_run:
+        async for node in agent_run:
+            if Agent.is_call_tools_node(node):
+                async with node.stream(agent_run.ctx) as event_stream:
+                    async for event in event_stream:
+                        event_parts.append(event)
+
+    assert event_parts == snapshot(
+        [
+            FunctionToolCallEvent(
+                part=ToolCallPart(
+                    tool_name='final_result',
+                    args={'bad_value': 'invalid'},
+                    tool_call_id=IsStr(),
+                ),
+            ),
+            FunctionToolResultEvent(
+                result=ToolReturnPart(
+                    tool_name='final_result',
+                    content='Output tool not used - result failed validation.',
+                    tool_call_id=IsStr(),
+                    timestamp=IsNow(tz=timezone.utc),
+                ),
+                tool_call_id=IsStr(),
+            ),
+        ]
+    )