From 1d2a54713bfae65873662f5a2b0255eec8852357 Mon Sep 17 00:00:00 2001 From: Danipulok Date: Sat, 22 Nov 2025 23:49:35 +0200 Subject: [PATCH 1/7] Make `end_strategy` also work for output tools, not just tools --- docs/output.md | 47 +++++++++ docs/tools-advanced.md | 8 ++ pydantic_ai_slim/pydantic_ai/_agent_graph.py | 38 +++++++- .../pydantic_ai/agent/__init__.py | 8 +- tests/test_agent.py | 96 ++++++++++++++++++- 5 files changed, 190 insertions(+), 7 deletions(-) diff --git a/docs/output.md b/docs/output.md index 182a753944..a9e975ffec 100644 --- a/docs/output.md +++ b/docs/output.md @@ -306,6 +306,53 @@ print(repr(result.output)) _(This example is complete, it can be run "as is")_ +##### Handling Multiple Output Tool Calls + +When a model calls multiple output tools in the same response (for example, when you have multiple output types in a union or list), the agent's `end_strategy` parameter controls whether all output tool functions are executed or only the first one: + +- `'early'` (default): Only the first output tool is executed, and additional output tool calls are skipped once a final result is found. This is the default behavior. +- `'exhaustive'`: All output tool functions are executed, even after a final result is found. The first tool's result is still used as the final output. + +This parameter also applies to [function tools](tools.md), not just output tools. + +```python {title="exhaustive_output.py"} +from pydantic import BaseModel +from pydantic_ai import Agent, ToolOutput + +calls_made: list[str] = [] + +class ResultA(BaseModel): + value: str + +class ResultB(BaseModel): + value: str + +def process_a(result: ResultA) -> ResultA: + calls_made.append('A') + return result + +def process_b(result: ResultB) -> ResultB: + calls_made.append('B') + return result + +# With 'exhaustive' strategy, both functions will be called +agent = Agent( + 'openai:gpt-5', + output_type=[ + ToolOutput(process_a, name='return_a'), + ToolOutput(process_b, name='return_b'), + ], + end_strategy='exhaustive', # (1)! +) + +# If the model calls both output tools, both process_a and process_b will execute +# calls_made will be ['A', 'B'] +``` + +1. Setting `end_strategy='exhaustive'` ensures all output tool functions are executed, which can be useful for side effects like logging, metrics, or validation. + +_(This example is complete, it can be run "as is")_ + #### Native Output Native Output mode uses a model's native "Structured Outputs" feature (aka "JSON Schema response format"), where the model is forced to only output text matching the provided JSON schema. Note that this is not supported by all models, and sometimes comes with restrictions. For example, Anthropic does not support this at all, and Gemini cannot use tools at the same time as structured output, and attempting to do so will result in an error. diff --git a/docs/tools-advanced.md b/docs/tools-advanced.md index f01a243c11..b257764b85 100644 --- a/docs/tools-advanced.md +++ b/docs/tools-advanced.md @@ -378,6 +378,14 @@ If a tool requires sequential/serial execution, you can pass the [`sequential`][ Async functions are run on the event loop, while sync functions are offloaded to threads. To get the best performance, _always_ use an async function _unless_ you're doing blocking I/O (and there's no way to use a non-blocking library instead) or CPU-bound work (like `numpy` or `scikit-learn` operations), so that simple functions are not offloaded to threads unnecessarily. +!!! note "Handling tool calls after a final result" + When a model returns multiple tool calls including an [output tool](output.md) (which produces a final result), the agent's `end_strategy` parameter controls whether remaining function tools are executed: + + - `'early'` (default): Function tools are skipped once a final result is found + - `'exhaustive'`: All function tools are executed even after a final result is found + + This is useful when function tools have side effects (like logging or metrics) that should always execute. See [Handling Multiple Output Tool Calls](output.md#handling-multiple-output-tool-calls) for more details. + !!! note "Limiting tool executions" You can cap tool executions within a run using [`UsageLimits(tool_calls_limit=...)`](agents.md#usage-limits). The counter increments only after a successful tool invocation. Output tools (used for [structured output](output.md)) are not counted in the `tool_calls` metric. diff --git a/pydantic_ai_slim/pydantic_ai/_agent_graph.py b/pydantic_ai_slim/pydantic_ai/_agent_graph.py index c973befc70..6f3e25a2be 100644 --- a/pydantic_ai_slim/pydantic_ai/_agent_graph.py +++ b/pydantic_ai_slim/pydantic_ai/_agent_graph.py @@ -61,8 +61,10 @@ EndStrategy = Literal['early', 'exhaustive'] """The strategy for handling multiple tool calls when a final result is found. -- `'early'`: Stop processing other tool calls once a final result is found -- `'exhaustive'`: Process all tool calls even after finding a final result +- `'early'`: Stop processing other tool calls (both function tools and output tools) once a final result is found +- `'exhaustive'`: Process all tool calls (both function tools and output tools) even after finding a final result + +This applies to both function tools and output tools. """ DepsT = TypeVar('DepsT') OutputT = TypeVar('OutputT') @@ -831,7 +833,9 @@ async def process_tool_calls( # noqa: C901 content='Final result processed.', tool_call_id=call.tool_call_id, ) - else: + output_parts.append(part) + # With early strategy, execute only the first output tool + elif ctx.deps.end_strategy == 'early': yield _messages.FunctionToolCallEvent(call) part = _messages.ToolReturnPart( tool_name=call.tool_name, @@ -839,8 +843,32 @@ async def process_tool_calls( # noqa: C901 tool_call_id=call.tool_call_id, ) yield _messages.FunctionToolResultEvent(part) - - output_parts.append(part) + output_parts.append(part) + # With exhaustive strategy, execute all output tools + elif ctx.deps.end_strategy == 'exhaustive': + try: + 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) + else: + assert_never(ctx.deps.end_strategy) else: try: result_data = await tool_manager.handle_call(call) diff --git a/pydantic_ai_slim/pydantic_ai/agent/__init__.py b/pydantic_ai_slim/pydantic_ai/agent/__init__.py index 4cd353b44a..e21af557d9 100644 --- a/pydantic_ai_slim/pydantic_ai/agent/__init__.py +++ b/pydantic_ai_slim/pydantic_ai/agent/__init__.py @@ -115,7 +115,13 @@ class Agent(AbstractAgent[AgentDepsT, OutputDataT]): _name: str | None end_strategy: EndStrategy - """Strategy for handling tool calls when a final result is found.""" + """Strategy for handling tool calls when a final result is found. + + - `'early'`: Stop processing other tool calls once a final result is found (default) + - `'exhaustive'`: Process all tool calls even after finding a final result + + This applies to both function tools and output tools. + """ model_settings: ModelSettings | None """Optional model request settings to use for this agents's runs, by default. diff --git a/tests/test_agent.py b/tests/test_agent.py index 8cc6b8b38c..9e0e230ddf 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -3132,7 +3132,7 @@ def deferred_tool(x: int) -> int: # pragma: no cover ), ToolReturnPart( tool_name='final_result', - content='Output tool not used - a final result was already processed.', + content='Final result processed.', tool_call_id=IsStr(), timestamp=IsNow(tz=timezone.utc), ), @@ -3163,6 +3163,100 @@ def deferred_tool(x: int) -> int: # pragma: no cover ] ) + def test_exhaustive_strategy_calls_all_output_tools(self): + """Test that 'exhaustive' strategy executes all output tool functions.""" + output_tools_called: list[str] = [] + + class FirstOutput(BaseModel): + value: str + + class SecondOutput(BaseModel): + value: str + + def process_first(output: FirstOutput) -> FirstOutput: + """Process first output.""" + output_tools_called.append('first') + return output + + def process_second(output: SecondOutput) -> SecondOutput: + """Process second output.""" + output_tools_called.append('second') + return output + + def return_model(_: list[ModelMessage], info: AgentInfo) -> ModelResponse: + assert info.output_tools is not None + return ModelResponse( + parts=[ + ToolCallPart('first_output', {'value': 'first'}), + ToolCallPart('second_output', {'value': 'second'}), + ], + ) + + agent = Agent( + FunctionModel(return_model), + output_type=[ + ToolOutput(process_first, name='first_output'), + ToolOutput(process_second, name='second_output'), + ], + end_strategy='exhaustive', + ) + + result = agent.run_sync('test exhaustive output tools') + + # Verify the result came from the first output tool + assert isinstance(result.output, FirstOutput) + assert result.output.value == 'first' + + # Verify both output tools were called + assert output_tools_called == ['first', 'second'] + + def test_early_strategy_does_not_call_additional_output_tools(self): + """Test that 'early' strategy does not execute additional output tool functions.""" + output_tools_called: list[str] = [] + + class FirstOutput(BaseModel): + value: str + + class SecondOutput(BaseModel): + value: str + + def process_first(output: FirstOutput) -> FirstOutput: + """Process first output.""" + output_tools_called.append('first') + return output + + def process_second(output: SecondOutput) -> SecondOutput: # pragma: no cover + """Process second output.""" + output_tools_called.append('second') + return output + + def return_model(_: list[ModelMessage], info: AgentInfo) -> ModelResponse: + assert info.output_tools is not None + return ModelResponse( + parts=[ + ToolCallPart('first_output', {'value': 'first'}), + ToolCallPart('second_output', {'value': 'second'}), + ], + ) + + agent = Agent( + FunctionModel(return_model), + output_type=[ + ToolOutput(process_first, name='first_output'), + ToolOutput(process_second, name='second_output'), + ], + end_strategy='early', + ) + + result = agent.run_sync('test early output tools') + + # Verify the result came from the first output tool + assert isinstance(result.output, FirstOutput) + assert result.output.value == 'first' + + # Verify only the first output tool was called + assert output_tools_called == ['first'] + def test_early_strategy_with_final_result_in_middle(self): """Test that 'early' strategy stops at first final result, regardless of position.""" tool_called = [] From b3371e5b3873e3251373600c9fe9ce47a4a39009 Mon Sep 17 00:00:00 2001 From: Danipulok Date: Sun, 23 Nov 2025 00:40:48 +0200 Subject: [PATCH 2/7] Fix tests --- tests/test_streaming.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 0c6a46f3c0..ade007da33 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -943,7 +943,7 @@ def another_tool(y: int) -> int: ), ToolReturnPart( tool_name='final_result', - content='Output tool not used - a final result was already processed.', + content='Final result processed.', timestamp=IsNow(tz=timezone.utc), tool_call_id=IsStr(), ), From c5bfc1b53f9f1dca6a122819bf77a49b74988bba Mon Sep 17 00:00:00 2001 From: Danipulok Date: Sun, 23 Nov 2025 01:08:45 +0200 Subject: [PATCH 3/7] Fix lints --- docs/output.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/output.md b/docs/output.md index a9e975ffec..2fa3719b76 100644 --- a/docs/output.md +++ b/docs/output.md @@ -317,6 +317,7 @@ This parameter also applies to [function tools](tools.md), not just output tools ```python {title="exhaustive_output.py"} from pydantic import BaseModel + from pydantic_ai import Agent, ToolOutput calls_made: list[str] = [] From 7856ce779bc15b27b647e7bae8522aa0e5d55d7a Mon Sep 17 00:00:00 2001 From: Danipulok Date: Sun, 23 Nov 2025 01:45:49 +0200 Subject: [PATCH 4/7] Add missing coverage --- tests/test_streaming.py | 43 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/tests/test_streaming.py b/tests/test_streaming.py index ade007da33..c0b95a4e63 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -2191,3 +2191,46 @@ async def text_stream(_messages: list[ModelMessage], agent_info: AgentInfo) -> A chunks.append(response_data) assert chunks == snapshot([[1], [1, 2, 3, 4], [1, 2, 3, 4], [1, 2, 3, 4]]) + + +async def test_exhaustive_strategy_multiple_final_results_with_retry(): + async def sf(_: list[ModelMessage], info: AgentInfo) -> AsyncIterator[str | DeltaToolCalls]: + # Return two final_result calls in one response + # The second one will fail validation + yield {1: DeltaToolCall('final_result', '{"value": "first"}')} + yield {2: DeltaToolCall('final_result', '{"value": 123}')} # Invalid - not a string + + agent = Agent(FunctionModel(stream_function=sf), output_type=OutputType, end_strategy='exhaustive') + + async with agent.run_stream('test retry') as result: + response = await result.get_output() + # Should use the first final result + assert response.value == snapshot('first') + + messages = result.all_messages() + # Check that we got a retry prompt for the second final_result + retry_parts = [ + p for m in messages if isinstance(m, ModelRequest) for p in m.parts if isinstance(p, RetryPromptPart) + ] + assert len(retry_parts) >= 1 + # The retry should mention validation error + assert any('value' in str(p.content).lower() for p in retry_parts) + + +async def test_exhaustive_strategy_final_result_unexpected_behavior(): + class CustomOutputType(BaseModel): + value: str + + async def sf(_: list[ModelMessage], info: AgentInfo) -> AsyncIterator[str | DeltaToolCalls]: + # Return two final_result calls where second one is malformed + yield {1: DeltaToolCall('final_result', '{"value": "first"}')} + yield {2: DeltaToolCall('final_result', 'not valid json')} # Malformed JSON + + agent = Agent( + FunctionModel(stream_function=sf), output_type=CustomOutputType, end_strategy='exhaustive', output_retries=0 + ) + + # Should raise because the second final_result has invalid JSON + with pytest.raises(UnexpectedModelBehavior): + async with agent.run_stream('test') as result: + await result.get_output() From 52d6a4c4dea27f61e8ee30bd9fa358389e5d0b96 Mon Sep 17 00:00:00 2001 From: Danipulok Date: Tue, 25 Nov 2025 00:17:29 +0200 Subject: [PATCH 5/7] Update tests to use snapshot --- tests/test_agent.py | 74 +++++++++++++++++++++++++++++++++++++++++ tests/test_streaming.py | 64 +++++++++++++++++++++++++++-------- 2 files changed, 124 insertions(+), 14 deletions(-) diff --git a/tests/test_agent.py b/tests/test_agent.py index 9e0e230ddf..c1df00b806 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -3210,6 +3210,43 @@ def return_model(_: list[ModelMessage], info: AgentInfo) -> ModelResponse: # Verify both output tools were called assert output_tools_called == ['first', 'second'] + # Verify we got tool returns in the correct order + assert result.all_messages() == snapshot( + [ + ModelRequest( + parts=[UserPromptPart(content='test exhaustive output tools', timestamp=IsNow(tz=timezone.utc))], + run_id=IsStr(), + ), + ModelResponse( + parts=[ + ToolCallPart(tool_name='first_output', args={'value': 'first'}, tool_call_id=IsStr()), + ToolCallPart(tool_name='second_output', args={'value': 'second'}, tool_call_id=IsStr()), + ], + usage=RequestUsage(input_tokens=54, output_tokens=10), + model_name='function:return_model:', + timestamp=IsNow(tz=timezone.utc), + run_id=IsStr(), + ), + ModelRequest( + parts=[ + ToolReturnPart( + tool_name='first_output', + content='Final result processed.', + tool_call_id=IsStr(), + timestamp=IsNow(tz=timezone.utc), + ), + ToolReturnPart( + tool_name='second_output', + content='Final result processed.', + tool_call_id=IsStr(), + timestamp=IsNow(tz=timezone.utc), + ), + ], + run_id=IsStr(), + ), + ] + ) + def test_early_strategy_does_not_call_additional_output_tools(self): """Test that 'early' strategy does not execute additional output tool functions.""" output_tools_called: list[str] = [] @@ -3257,6 +3294,43 @@ def return_model(_: list[ModelMessage], info: AgentInfo) -> ModelResponse: # Verify only the first output tool was called assert output_tools_called == ['first'] + # Verify we got tool returns in the correct order + assert result.all_messages() == snapshot( + [ + ModelRequest( + parts=[UserPromptPart(content='test early output tools', timestamp=IsNow(tz=timezone.utc))], + run_id=IsStr(), + ), + ModelResponse( + parts=[ + ToolCallPart(tool_name='first_output', args={'value': 'first'}, tool_call_id=IsStr()), + ToolCallPart(tool_name='second_output', args={'value': 'second'}, tool_call_id=IsStr()), + ], + usage=RequestUsage(input_tokens=54, output_tokens=10), + model_name='function:return_model:', + timestamp=IsNow(tz=timezone.utc), + run_id=IsStr(), + ), + ModelRequest( + parts=[ + ToolReturnPart( + tool_name='first_output', + content='Final result processed.', + tool_call_id=IsStr(), + timestamp=IsNow(tz=timezone.utc), + ), + ToolReturnPart( + tool_name='second_output', + content='Output tool not used - a final result was already processed.', + tool_call_id=IsStr(), + timestamp=IsNow(tz=timezone.utc), + ), + ], + run_id=IsStr(), + ), + ] + ) + def test_early_strategy_with_final_result_in_middle(self): """Test that 'early' strategy stops at first final result, regardless of position.""" tool_called = [] diff --git a/tests/test_streaming.py b/tests/test_streaming.py index c0b95a4e63..b1d28ca9e8 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -12,6 +12,7 @@ import pytest from inline_snapshot import snapshot from pydantic import BaseModel +from pydantic_core import ErrorDetails from pydantic_ai import ( Agent, @@ -1585,12 +1586,12 @@ def call_final_result_with_bad_data(messages: list[ModelMessage], info: AgentInf FunctionToolResultEvent( result=RetryPromptPart( content=[ - { - 'type': 'missing', - 'loc': ('value',), - 'msg': 'Field required', - 'input': {'bad_value': 'invalid'}, - } + ErrorDetails( + type='missing', + loc=('value',), + msg='Field required', + input={'bad_value': 'invalid'}, + ), ], tool_name='final_result', tool_call_id=IsStr(), @@ -2207,14 +2208,49 @@ async def sf(_: list[ModelMessage], info: AgentInfo) -> AsyncIterator[str | Delt # Should use the first final result assert response.value == snapshot('first') - messages = result.all_messages() - # Check that we got a retry prompt for the second final_result - retry_parts = [ - p for m in messages if isinstance(m, ModelRequest) for p in m.parts if isinstance(p, RetryPromptPart) - ] - assert len(retry_parts) >= 1 - # The retry should mention validation error - assert any('value' in str(p.content).lower() for p in retry_parts) + # Verify we got the expected message flow with retry for the second final_result + assert result.all_messages() == snapshot( + [ + ModelRequest( + parts=[UserPromptPart(content='test retry', timestamp=IsNow(tz=timezone.utc))], + run_id=IsStr(), + ), + ModelResponse( + parts=[ + ToolCallPart(tool_name='final_result', args='{"value": "first"}', tool_call_id=IsStr()), + ToolCallPart(tool_name='final_result', args='{"value": 123}', tool_call_id=IsStr()), + ], + usage=RequestUsage(input_tokens=50, output_tokens=7), + model_name='function::sf', + timestamp=IsNow(tz=timezone.utc), + run_id=IsStr(), + ), + ModelRequest( + parts=[ + ToolReturnPart( + tool_name='final_result', + content='Final result processed.', + tool_call_id=IsStr(), + timestamp=IsNow(tz=timezone.utc), + ), + RetryPromptPart( + content=[ + ErrorDetails( + type='string_type', + loc=('value',), + msg='Input should be a valid string', + input=123, + ) + ], + tool_name='final_result', + tool_call_id=IsStr(), + timestamp=IsNow(tz=timezone.utc), + ), + ], + run_id=IsStr(), + ), + ] + ) async def test_exhaustive_strategy_final_result_unexpected_behavior(): From 9a273601bc8ad29d74e45a9d3228a8a4c8f6a403 Mon Sep 17 00:00:00 2001 From: Danipulok Date: Tue, 25 Nov 2025 00:30:51 +0200 Subject: [PATCH 6/7] Refactor duplicated code --- pydantic_ai_slim/pydantic_ai/_agent_graph.py | 73 ++++++++------------ 1 file changed, 28 insertions(+), 45 deletions(-) diff --git a/pydantic_ai_slim/pydantic_ai/_agent_graph.py b/pydantic_ai_slim/pydantic_ai/_agent_graph.py index 6f3e25a2be..4f547c72a7 100644 --- a/pydantic_ai_slim/pydantic_ai/_agent_graph.py +++ b/pydantic_ai_slim/pydantic_ai/_agent_graph.py @@ -826,50 +826,27 @@ async def process_tool_calls( # noqa: C901 # 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, - ) - output_parts.append(part) - # With early strategy, execute only the first output tool - elif ctx.deps.end_strategy == 'early': - 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) - # With exhaustive strategy, execute all output tools - elif ctx.deps.end_strategy == 'exhaustive': - try: - 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) - else: - assert_never(ctx.deps.end_strategy) - else: + # In case we got two tool calls with the same ID + if final_result and 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, + ) + output_parts.append(part) + # Early strategy is chosen and final result is already set + elif ctx.deps.end_strategy == 'early' and final_result: + 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) + # Early strategy is chosen and final result is not yet set + # Or exhaustive strategy is chosen + elif (ctx.deps.end_strategy == 'early' and not final_result) or ctx.deps.end_strategy == 'exhaustive': try: result_data = await tool_manager.handle_call(call) except exceptions.UnexpectedModelBehavior as e: @@ -891,7 +868,13 @@ async def process_tool_calls( # noqa: C901 tool_call_id=call.tool_call_id, ) output_parts.append(part) - final_result = result.FinalResult(result_data, call.tool_name, call.tool_call_id) + + # Even in exhaustive mode, use the first output tool's result as the final result + if not final_result: + final_result = result.FinalResult(result_data, call.tool_name, call.tool_call_id) + # Unknown strategy or invalid state + else: + assert_never(ctx.deps.end_strategy) # Then, we handle function tool calls calls_to_run: list[_messages.ToolCallPart] = [] From 4a1a34e4e141a95a344af1965b27d82e226bcbce Mon Sep 17 00:00:00 2001 From: Danipulok Date: Tue, 25 Nov 2025 02:52:34 +0200 Subject: [PATCH 7/7] Fix typecheck errors --- pydantic_ai_slim/pydantic_ai/_agent_graph.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pydantic_ai_slim/pydantic_ai/_agent_graph.py b/pydantic_ai_slim/pydantic_ai/_agent_graph.py index 4f547c72a7..40e2a31040 100644 --- a/pydantic_ai_slim/pydantic_ai/_agent_graph.py +++ b/pydantic_ai_slim/pydantic_ai/_agent_graph.py @@ -874,7 +874,9 @@ async def process_tool_calls( # noqa: C901 final_result = result.FinalResult(result_data, call.tool_name, call.tool_call_id) # Unknown strategy or invalid state else: - assert_never(ctx.deps.end_strategy) + assert False, ( + f'Unexpected state: end_strategy={ctx.deps.end_strategy!r}, final_result={final_result!r};' + ) # pragma: no cover # Then, we handle function tool calls calls_to_run: list[_messages.ToolCallPart] = []