Support Agent(output_type=str | None) for optional output (#4042)

Co-authored-by: pavan <pavanmanikanta98@gmail.com>
Co-authored-by: David <64162682+dsfaccini@users.noreply.github.com>
Co-authored-by: Douwe Maan <douwe@pydantic.dev>
Co-authored-by: Douwe Maan <hi@douwe.me>

Author: 4 people

docs/output.md

@@ -675,6 +675,44 @@ Once upon a time, in a hidden underwater cave, lived a curious axolotl named Pip
 """
 ```
 
+## Optional output (allowing `None`) {#optional-output}
+
+Some agents perform their work entirely through tool calls and don't need to produce a final output — for example, an agent that updates a record via a tool and then stops. Certain models (notably [Anthropic](models/anthropic.md)) will return an empty response in this case, which by default causes Pydantic AI to retry until the model produces content.
+
+To instead treat an empty response as a successful run, include `None` in the `output_type`:
+
+```python {title="optional_output.py"}
+from pydantic_ai import Agent
+
+agent = Agent('anthropic:claude-opus-4-6', output_type=str | None)
+
+
+@agent.tool_plain
+def mark_task_done(task_id: int) -> str:
+    """Mark the task as done."""
+    return f'Task {task_id} marked done.'
+
+
+result = agent.run_sync('Mark task 1 as done, then stop without saying anything.')
+print(result.output)
+#> None
+```
+
+When the model returns an empty response and `None` is an allowed output type, the agent will return `None` instead of retrying. [Output validator functions](#output-validator-functions) still run with `None` as the argument, so you can raise [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] to reject it if needed.
+
+`output_type=str | None` is the canonical case: it's handled as regular text output, and the **only** way the model signals `None` is by returning an empty response — there's no output tool or structured schema involved. This mirrors how plain `str` is already treated specially as free-form text output rather than a structured tool call.
+
+`None` is also supported in the other output modes, with an extra structured commit path in addition to (or in place of) the empty-response fallback:
+
+- **Bare unions including `None` that use tool mode** — e.g. `output_type=int | None`, `output_type=[int, float, None]`, or `output_type=[ToolOutput(Foo), None]`: a dedicated `final_result_NoneType` output tool is exposed alongside the other output tools, so the model can commit to `None` through a tool call. An empty model response is still also treated as `None`, as with `str | None`.
+- **Explicit output mode markers** — e.g. `output_type=ToolOutput(int | None)`, `output_type=NativeOutput([int, None])`, or `output_type=PromptedOutput([int, None])`: `None` is included as a branch of the structured schema the wrapper generates. The model commits by calling the tool with `null` (for `ToolOutput`) or by selecting the `NoneType` branch of the discriminated schema (for `NativeOutput`/`PromptedOutput`). An empty response is **not** accepted — once you've opted into an explicit structured output mode, the model is expected to commit through the schema.
+
+!!! note
+    `output_type=None` on its own is not valid — at least one other output type must be provided alongside `None`.
+
+!!! note
+    When using [`agent.run_stream()`][pydantic_ai.Agent.run_stream] with an optional output type, an empty model response has no intermediate values to yield, so [`stream_output()`][pydantic_ai.result.StreamedRunResult.stream_output] produces an empty iterator in this case. Use [`get_output()`][pydantic_ai.result.StreamedRunResult.get_output] to retrieve the final `None` value instead.
+
 ## Streamed Results
 
 There two main challenges with streamed results:

pydantic_ai_slim/pydantic_ai/_agent_graph.py

@@ -1028,6 +1028,19 @@ async def _run_stream() -> AsyncIterator[_messages.HandleResponseEvent]:  # noqa
 
                         raise exceptions.ContentFilterError(message, body=body)
 
+                    # If the output type allows None, an empty response is a valid result.
+                    if is_empty and output_schema.allows_none:
+                        run_context = _build_output_run_context(ctx)
+                        try:
+                            result_data = await _run_output_validators(ctx, cast(NodeRunEndT, None), run_context)
+                            self._next_node = self._handle_final_result(ctx, result.FinalResult(result_data), [])
+                        except ToolRetryError as e:
+                            ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
+                            self._next_node = ModelRequestNode[DepsT, NodeRunEndT](
+                                _messages.ModelRequest(parts=[e.tool_retry])
+                            )
+                        return
+
                     # Try to recover text from a previous model response.
                     # This handles the case where the model returned text alongside tool calls
                     # (so the text was discarded in favor of executing the tools) and subsequently

pydantic_ai_slim/pydantic_ai/_output.py

@@ -6,6 +6,7 @@
 from collections.abc import Awaitable, Callable, Sequence
 from copy import copy
 from dataclasses import dataclass, field, replace
+from types import NoneType
 from typing import TYPE_CHECKING, Any, Generic, Literal, cast, overload
 
 from pydantic import Json, TypeAdapter, ValidationError
@@ -213,6 +214,7 @@ async def validate(
 
 @dataclass(kw_only=True)
 class OutputSchema(ABC, Generic[OutputDataT]):
+    allows_none: bool
     text_processor: BaseOutputProcessor[OutputDataT] | None = None
     toolset: OutputToolset[Any] | None = None
     object_def: OutputObjectDefinition | None = None
@@ -239,6 +241,13 @@ def build(  # noqa: C901
         """Build an OutputSchema dataclass from an output type."""
         outputs = _flatten_output_spec(output_spec)
 
+        # `str | None` produces NoneType (the class) via get_union_args; bare `None` value produces None itself
+        allows_none = NoneType in outputs or None in outputs
+        if allows_none:
+            outputs = [output for output in outputs if output is not NoneType and output is not None]
+            if len(outputs) == 0:
+                raise UserError('At least one output type must be provided other than `None`.')
+
         allows_deferred_tools = DeferredToolRequests in outputs
         if allows_deferred_tools:
             outputs = [output for output in outputs if output is not DeferredToolRequests]
@@ -274,6 +283,7 @@ def build(  # noqa: C901
                 ),
                 allows_deferred_tools=allows_deferred_tools,
                 allows_image=allows_image,
+                allows_none=allows_none,
             )
         elif output := next((output for output in outputs if isinstance(output, PromptedOutput)), None):  # pyright: ignore[reportUnknownVariableType,reportUnknownArgumentType]
             if len(outputs) > 1:
@@ -299,6 +309,7 @@ def build(  # noqa: C901
                 ),
                 allows_deferred_tools=allows_deferred_tools,
                 allows_image=allows_image,
+                allows_none=allows_none,
             )
 
         text_outputs: Sequence[type[str] | TextOutput[OutputDataT]] = []
@@ -320,6 +331,12 @@ def build(  # noqa: C901
             else:
                 other_outputs.append(output)
 
+        # If `None` is allowed and we're building output tools, expose `NoneType` as its own
+        # output tool so the model can commit to `None` through the structured schema alongside
+        # any other output types, matching how the model would pick between them.
+        if allows_none and (tool_outputs or other_outputs):
+            other_outputs.append(cast(OutputTypeOrFunction[OutputDataT], NoneType))
+
         toolset = OutputToolset.build(tool_outputs + other_outputs, name=name, description=description, strict=strict)
 
         text_processor: BaseOutputProcessor[OutputDataT] | None = None
@@ -340,17 +357,22 @@ def build(  # noqa: C901
                     text_processor=text_processor,
                     allows_deferred_tools=allows_deferred_tools,
                     allows_image=allows_image,
+                    allows_none=allows_none,
                 )
             else:
                 return TextOutputSchema(
                     text_processor=text_processor,
                     allows_deferred_tools=allows_deferred_tools,
                     allows_image=allows_image,
+                    allows_none=allows_none,
                 )
 
         if len(tool_outputs) > 0:
             return ToolOutputSchema(
-                toolset=toolset, allows_deferred_tools=allows_deferred_tools, allows_image=allows_image
+                toolset=toolset,
+                allows_deferred_tools=allows_deferred_tools,
+                allows_image=allows_image,
+                allows_none=allows_none,
             )
 
         if len(other_outputs) > 0:
@@ -359,10 +381,11 @@ def build(  # noqa: C901
                 toolset=toolset,
                 allows_deferred_tools=allows_deferred_tools,
                 allows_image=allows_image,
+                allows_none=allows_none,
             )
 
         if allows_image:
-            return ImageOutputSchema(allows_deferred_tools=allows_deferred_tools)
+            return ImageOutputSchema(allows_deferred_tools=allows_deferred_tools, allows_none=allows_none)
 
         raise UserError('At least one output type must be provided.')
 
@@ -390,6 +413,7 @@ def __init__(
         toolset: OutputToolset[Any] | None,
         allows_deferred_tools: bool,
         allows_image: bool,
+        allows_none: bool,
     ):
         # We set a toolset here as they're checked for name conflicts with other toolsets in the Agent constructor.
         # At that point we may not know yet what output mode we're going to use if no model was provided or it was deferred until agent.run time,
@@ -400,6 +424,7 @@ def __init__(
             text_processor=processor,
             allows_deferred_tools=allows_deferred_tools,
             allows_image=allows_image,
+            allows_none=allows_none,
         )
         self.processor = processor
 
@@ -416,11 +441,13 @@ def __init__(
         text_processor: TextOutputProcessor[OutputDataT],
         allows_deferred_tools: bool,
         allows_image: bool,
+        allows_none: bool,
     ):
         super().__init__(
             text_processor=text_processor,
             allows_deferred_tools=allows_deferred_tools,
             allows_image=allows_image,
+            allows_none=allows_none,
         )
 
     @property
@@ -429,8 +456,8 @@ def mode(self) -> OutputMode:
 
 
 class ImageOutputSchema(OutputSchema[OutputDataT]):
-    def __init__(self, *, allows_deferred_tools: bool):
-        super().__init__(allows_deferred_tools=allows_deferred_tools, allows_image=True)
+    def __init__(self, *, allows_deferred_tools: bool, allows_none: bool):
+        super().__init__(allows_deferred_tools=allows_deferred_tools, allows_image=True, allows_none=allows_none)
 
     @property
     def mode(self) -> OutputMode:
@@ -449,12 +476,14 @@ def __init__(
         processor: BaseObjectOutputProcessor[OutputDataT],
         allows_deferred_tools: bool,
         allows_image: bool,
+        allows_none: bool,
     ):
         super().__init__(
             text_processor=processor,
             object_def=processor.object_def,
             allows_deferred_tools=allows_deferred_tools,
             allows_image=allows_image,
+            allows_none=allows_none,
         )
         self.processor = processor
         self.template = template
@@ -496,12 +525,14 @@ def __init__(
         text_processor: BaseOutputProcessor[OutputDataT] | None = None,
         allows_deferred_tools: bool,
         allows_image: bool,
+        allows_none: bool,
     ):
         super().__init__(
             toolset=toolset,
             allows_deferred_tools=allows_deferred_tools,
             text_processor=text_processor,
             allows_image=allows_image,
+            allows_none=allows_none,
         )
 
     @property