Add missing Vercel AI SDK v6 tool approval part types

docs/deferred-tools.md

@@ -169,6 +169,7 @@ print(result.all_messages())
                 content='Deleting files is not allowed',
                 tool_call_id='delete_file',
                 timestamp=datetime.datetime(...),
+                outcome='denied',
             ),
             UserPromptPart(
                 content='Now create a backup of README.md',

pydantic_ai_slim/pydantic_ai/_agent_graph.py

@@ -1340,6 +1340,7 @@ async def _call_tool(
                 tool_name=call.tool_name,
                 content=tool_call_result.message,
                 tool_call_id=call.tool_call_id,
+                outcome='denied',
             ), None
         elif isinstance(tool_call_result, exceptions.ModelRetry):
             m = _messages.RetryPromptPart(

pydantic_ai_slim/pydantic_ai/messages.py

@@ -1024,6 +1024,14 @@ class BaseToolReturnPart:
     timestamp: datetime = field(default_factory=_now_utc)
     """The timestamp, when the tool returned."""
 
+    outcome: Literal['success', 'failed', 'denied'] = 'success'
+    """The outcome of the tool call.
+
+    - `'success'`: The tool executed successfully.
+    - `'failed'`: The tool raised an error during execution.
+    - `'denied'`: The tool call was denied by the approval mechanism.
+    """
+
     def model_response_str(self) -> str:
         """Return a string representation of the content for the model."""
         if isinstance(self.content, str):

pydantic_ai_slim/pydantic_ai/ui/__init__.py

@@ -1,9 +1,13 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from ._adapter import StateDeps, StateHandler, UIAdapter
 from ._event_stream import SSE_CONTENT_TYPE, NativeEvent, OnCompleteFunc, UIEventStream
 from ._messages_builder import MessagesBuilder
-from ._web import DEFAULT_HTML_URL
+
+if TYPE_CHECKING:
+    from ._web import DEFAULT_HTML_URL
 
 __all__ = [
     'UIAdapter',
@@ -16,3 +20,11 @@
     'MessagesBuilder',
     'DEFAULT_HTML_URL',
 ]
+
+
+def __getattr__(name: str) -> object:
+    if name == 'DEFAULT_HTML_URL':
+        from ._web import DEFAULT_HTML_URL
+
+        return DEFAULT_HTML_URL
+    raise AttributeError(f'module {__name__!r} has no attribute {name!r}')

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/_adapter.py

@@ -58,8 +58,10 @@
     SourceUrlUIPart,
     StepStartUIPart,
     TextUIPart,
+    ToolApprovalResponded,
     ToolInputAvailablePart,
     ToolOutputAvailablePart,
+    ToolOutputDeniedPart,
     ToolOutputErrorPart,
     ToolUIPart,
     UIMessage,
@@ -343,7 +345,9 @@ def load_messages(cls, messages: Sequence[UIMessage]) -> list[ModelMessage]:  #
                             # The call and return metadata are combined in the output part.
                             # So we extract and return them to the respective parts
                             call_meta = return_meta = {}
-                            has_tool_output = isinstance(part, (ToolOutputAvailablePart, ToolOutputErrorPart))
+                            has_tool_output = isinstance(
+                                part, (ToolOutputAvailablePart, ToolOutputErrorPart, ToolOutputDeniedPart)
+                            )
 
                             if has_tool_output:
                                 call_meta, return_meta = cls._load_builtin_tool_meta(provider_meta)
@@ -360,18 +364,23 @@ def load_messages(cls, messages: Sequence[UIMessage]) -> list[ModelMessage]:  #
                             )
 
                             if has_tool_output:
-                                output: Any | None = None
-                                if isinstance(part, ToolOutputAvailablePart):
-                                    output = part.output
-                                elif isinstance(part, ToolOutputErrorPart):  # pragma: no branch
-                                    output = {'error_text': part.error_text, 'is_error': True}
+                                if isinstance(part, ToolOutputErrorPart):
+                                    output: Any = part.error_text
+                                    outcome: Literal['success', 'failed', 'denied'] = 'failed'
+                                elif isinstance(part, ToolOutputDeniedPart):
+                                    output = _denial_reason(part)
+                                    outcome = 'denied'
+                                else:
+                                    output = part.output if isinstance(part, ToolOutputAvailablePart) else None
+                                    outcome = 'success'
                                 builder.add(
                                     BuiltinToolReturnPart(
                                         tool_name=tool_name,
                                         tool_call_id=tool_call_id,
                                         content=output,
                                         provider_name=return_meta.get('provider_name') or provider_name,
                                         provider_details=return_meta.get('provider_details') or provider_details,
+                                        outcome=outcome,
                                     )
                                 )
                         else:
@@ -392,8 +401,20 @@ def load_messages(cls, messages: Sequence[UIMessage]) -> list[ModelMessage]:  #
                                 )
                             elif part.state == 'output-error':
                                 builder.add(
-                                    RetryPromptPart(
-                                        tool_name=tool_name, tool_call_id=tool_call_id, content=part.error_text
+                                    ToolReturnPart(
+                                        tool_name=tool_name,
+                                        tool_call_id=tool_call_id,
+                                        content=part.error_text,
+                                        outcome='failed',
+                                    )
+                                )
+                            elif part.state == 'output-denied':
+                                builder.add(
+                                    ToolReturnPart(
+                                        tool_name=tool_name,
+                                        tool_call_id=tool_call_id,
+                                        content=_denial_reason(part),
+                                        outcome='denied',
                                     )
                                 )
                     elif isinstance(part, DataUIPart):  # pragma: no cover
@@ -522,20 +543,33 @@ def _dump_response_message(
                     )
                     combined_provider_meta = cls._dump_builtin_tool_meta(call_meta, return_meta)
 
-                    response_object = builtin_return.model_response_object()
-                    # These `is_error`/`error_text` fields are only present when the BuiltinToolReturnPart
-                    # was parsed from an incoming VercelAI request. We can't detect errors for other sources
-                    # until BuiltinToolReturnPart has standardized error fields (see https://github.com/pydantic/pydantic-ai/issues/3561).3
-                    if response_object.get('is_error') is True and (
-                        (error_text := response_object.get('error_text')) is not None
+                    if builtin_return.outcome == 'denied':
+                        ui_parts.append(
+                            ToolOutputDeniedPart(
+                                type=tool_name,
+                                tool_call_id=part.tool_call_id,
+                                input=_safe_args_as_dict(part),
+                                provider_executed=True,
+                                call_provider_metadata=combined_provider_meta,
+                                approval=ToolApprovalResponded(
+                                    id=str(uuid.uuid4()),
+                                    approved=False,
+                                    reason=builtin_return.model_response_str(),
+                                ),
+                            )
+                        )
+                    elif (
+                        builtin_return.outcome == 'failed'
+                        or builtin_return.model_response_object().get('is_error') is True
                     ):
+                        response_obj = builtin_return.model_response_object()
+                        error_text = response_obj.get('error_text', builtin_return.model_response_str())
                         ui_parts.append(
                             ToolOutputErrorPart(
                                 type=tool_name,
                                 tool_call_id=part.tool_call_id,
                                 input=_safe_args_as_dict(part),
                                 error_text=error_text,
-                                state='output-error',
                                 provider_executed=True,
                                 call_provider_metadata=combined_provider_meta,
                             )
@@ -547,7 +581,6 @@ def _dump_response_message(
                                 tool_call_id=part.tool_call_id,
                                 input=_safe_args_as_dict(part),
                                 output=tool_return_output(builtin_return),
-                                state='output-available',
                                 provider_executed=True,
                                 call_provider_metadata=combined_provider_meta,
                             )
@@ -561,58 +594,90 @@ def _dump_response_message(
                             type=tool_name,
                             tool_call_id=part.tool_call_id,
                             input=_safe_args_as_dict(part),
-                            state='input-available',
                             provider_executed=True,
                             call_provider_metadata=call_provider_metadata,
                         )
                     )
             elif isinstance(part, ToolCallPart):
-                tool_result = tool_results.get(part.tool_call_id)
-                call_provider_metadata = dump_provider_metadata(
-                    id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
-                )
-                tool_type = f'tool-{part.tool_name}'
+                ui_parts.extend(cls._dump_tool_call_part(part, tool_results))
+            else:
+                assert_never(part)
 
-                if isinstance(tool_result, ToolReturnPart):
-                    ui_parts.append(
-                        ToolOutputAvailablePart(
-                            type=tool_type,
-                            tool_call_id=part.tool_call_id,
-                            input=_safe_args_as_dict(part),
-                            output=tool_return_output(tool_result),
-                            state='output-available',
-                            provider_executed=False,
-                            call_provider_metadata=call_provider_metadata,
-                        )
-                    )
-                    # Check for Vercel AI chunks returned by tool calls via metadata.
-                    ui_parts.extend(_extract_metadata_ui_parts(tool_result))
-                elif isinstance(tool_result, RetryPromptPart):
-                    error_text = tool_result.model_response()
-                    ui_parts.append(
-                        ToolOutputErrorPart(
-                            type=tool_type,
-                            tool_call_id=part.tool_call_id,
-                            input=_safe_args_as_dict(part),
-                            error_text=error_text,
-                            state='output-error',
-                            provider_executed=False,
-                            call_provider_metadata=call_provider_metadata,
-                        )
+        return ui_parts
+
+    @staticmethod
+    def _dump_tool_call_part(
+        part: ToolCallPart, tool_results: dict[str, ToolReturnPart | RetryPromptPart]
+    ) -> list[UIMessagePart]:
+        """Convert a ToolCallPart (with optional result) into UIMessageParts."""
+        tool_result = tool_results.get(part.tool_call_id)
+        call_provider_metadata = dump_provider_metadata(
+            id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
+        )
+        tool_type = f'tool-{part.tool_name}'
+        ui_parts: list[UIMessagePart] = []
+
+        if isinstance(tool_result, ToolReturnPart):
+            if tool_result.outcome == 'denied':
+                ui_parts.append(
+                    ToolOutputDeniedPart(
+                        type=tool_type,
+                        tool_call_id=part.tool_call_id,
+                        input=_safe_args_as_dict(part),
+                        provider_executed=False,
+                        call_provider_metadata=call_provider_metadata,
+                        approval=ToolApprovalResponded(
+                            id=str(uuid.uuid4()),
+                            approved=False,
+                            reason=tool_result.model_response_str(),
+                        ),
                     )
-                else:
-                    ui_parts.append(
-                        ToolInputAvailablePart(
-                            type=tool_type,
-                            tool_call_id=part.tool_call_id,
-                            input=_safe_args_as_dict(part),
-                            state='input-available',
-                            provider_executed=False,
-                            call_provider_metadata=call_provider_metadata,
-                        )
+                )
+            elif tool_result.outcome == 'failed':
+                ui_parts.append(
+                    ToolOutputErrorPart(
+                        type=tool_type,
+                        tool_call_id=part.tool_call_id,
+                        input=_safe_args_as_dict(part),
+                        error_text=tool_result.model_response_str(),
+                        provider_executed=False,
+                        call_provider_metadata=call_provider_metadata,
                     )
+                )
             else:
-                assert_never(part)
+                ui_parts.append(
+                    ToolOutputAvailablePart(
+                        type=tool_type,
+                        tool_call_id=part.tool_call_id,
+                        input=_safe_args_as_dict(part),
+                        output=tool_return_output(tool_result),
+                        provider_executed=False,
+                        call_provider_metadata=call_provider_metadata,
+                    )
+                )
+            # Check for Vercel AI chunks returned by tool calls via metadata.
+            ui_parts.extend(_extract_metadata_ui_parts(tool_result))
+        elif isinstance(tool_result, RetryPromptPart):
+            ui_parts.append(
+                ToolOutputErrorPart(
+                    type=tool_type,
+                    tool_call_id=part.tool_call_id,
+                    input=_safe_args_as_dict(part),
+                    error_text=tool_result.model_response(),
+                    provider_executed=False,
+                    call_provider_metadata=call_provider_metadata,
+                )
+            )
+        else:
+            ui_parts.append(
+                ToolInputAvailablePart(
+                    type=tool_type,
+                    tool_call_id=part.tool_call_id,
+                    input=_safe_args_as_dict(part),
+                    provider_executed=False,
+                    call_provider_metadata=call_provider_metadata,
+                )
+            )
 
         return ui_parts
 
@@ -715,6 +780,13 @@ def _convert_user_prompt_part(part: UserPromptPart) -> list[UIMessagePart]:
     return ui_parts
 
 
+def _denial_reason(part: ToolUIPart | DynamicToolUIPart) -> str:
+    """Extract the denial reason from a tool part's approval, or return a default message."""
+    if isinstance(part.approval, ToolApprovalResponded) and part.approval.reason:
+        return part.approval.reason
+    return ToolDenied().message
+
+
 def _extract_metadata_ui_parts(tool_result: ToolReturnPart) -> list[UIMessagePart]:
     """Convert data-carrying chunks from tool metadata into UIMessageParts.
 

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/_event_stream.py

@@ -4,7 +4,6 @@
 
 from collections.abc import AsyncIterator, Mapping
 from dataclasses import KW_ONLY, dataclass
-from functools import cached_property
 from typing import Any, Literal
 from uuid import uuid4
 
@@ -29,7 +28,7 @@
 from ...run import AgentRunResultEvent
 from ...tools import AgentDepsT, DeferredToolRequests
 from .. import UIEventStream
-from ._utils import dump_provider_metadata, iter_metadata_chunks, iter_tool_approval_responses, tool_return_output
+from ._utils import dump_provider_metadata, iter_metadata_chunks, tool_return_output
 from .request_types import RequestData
 from .response_types import (
     BaseChunk,
@@ -87,15 +86,6 @@ class VercelAIEventStream(UIEventStream[RequestData, BaseChunk, AgentDepsT, Outp
     _step_started: bool = False
     _finish_reason: FinishReason = None
 
-    @cached_property
-    def _denied_tool_ids(self) -> set[str]:
-        """Get the set of tool_call_ids that were denied by the user."""
-        return {
-            tool_call_id
-            for tool_call_id, approval in iter_tool_approval_responses(self.run_input.messages)
-            if not approval.approved
-        }
-
     @property
     def response_headers(self) -> Mapping[str, str] | None:
         return VERCEL_AI_DSP_HEADERS
@@ -257,11 +247,16 @@ async def handle_builtin_tool_call_end(self, part: BuiltinToolCallPart) -> Async
         )
 
     async def handle_builtin_tool_return(self, part: BuiltinToolReturnPart) -> AsyncIterator[BaseChunk]:
-        yield ToolOutputAvailableChunk(
-            tool_call_id=part.tool_call_id,
-            output=tool_return_output(part),
-            provider_executed=True,
-        )
+        if self.sdk_version >= 6 and part.outcome == 'denied':
+            yield ToolOutputDeniedChunk(tool_call_id=part.tool_call_id)
+        elif part.outcome == 'failed':
+            yield ToolOutputErrorChunk(tool_call_id=part.tool_call_id, error_text=part.model_response_str())
+        else:
+            yield ToolOutputAvailableChunk(
+                tool_call_id=part.tool_call_id,
+                output=tool_return_output(part),
+                provider_executed=True,
+            )
 
     async def handle_file(self, part: FilePart) -> AsyncIterator[BaseChunk]:
         file = part.content
@@ -271,11 +266,12 @@ async def handle_function_tool_result(self, event: FunctionToolResultEvent) -> A
         part = event.result
         tool_call_id = part.tool_call_id
 
-        # Check if this tool was denied by the user (only when sdk_version >= 6)
-        if self.sdk_version >= 6 and tool_call_id in self._denied_tool_ids:
+        if self.sdk_version >= 6 and isinstance(part, ToolReturnPart) and part.outcome == 'denied':
             yield ToolOutputDeniedChunk(tool_call_id=tool_call_id)
         elif isinstance(part, RetryPromptPart):
             yield ToolOutputErrorChunk(tool_call_id=tool_call_id, error_text=part.model_response())
+        elif isinstance(part, ToolReturnPart) and part.outcome == 'failed':
+            yield ToolOutputErrorChunk(tool_call_id=tool_call_id, error_text=part.model_response_str())
         else:
             yield ToolOutputAvailableChunk(tool_call_id=tool_call_id, output=tool_return_output(part))