Add tool approval integration for Vercel AI adapter

docs/ui/vercel-ai.md

@@ -1,6 +1,6 @@
 # Vercel AI Data Stream Protocol
 
-Pydantic AI natively supports the [Vercel AI Data Stream Protocol](https://ai-sdk.dev/docs/ai-sdk-ui/stream-protocol#data-stream-protocol) to receive agent run input from, and stream events to, a [Vercel AI Elements](https://ai-sdk.dev/elements) frontend.
+Pydantic AI natively supports the [Vercel AI Data Stream Protocol](https://ai-sdk.dev/docs/ai-sdk-ui/stream-protocol#data-stream-protocol) to receive agent run input from, and stream events to, a frontend using [AI SDK UI](https://ai-sdk.dev/docs/ai-sdk-ui/overview) hooks like [`useChat`](https://ai-sdk.dev/docs/reference/ai-sdk-ui/use-chat). You can optionally use [AI Elements](https://ai-sdk.dev/elements) for pre-built UI components.
 
 !!! note
     By default, the adapter targets AI SDK v5 for backwards compatibility. To use features introduced in AI SDK v6, set `sdk_version=6` on the adapter.
@@ -123,3 +123,26 @@ async def search_docs(query: str) -> ToolReturn:
 
 !!! note
     Protocol-control chunks such as `StartChunk`, `FinishChunk`, `StartStepChunk`, or `FinishStepChunk` are automatically filtered out — only the four data-carrying chunk types listed above are forwarded to the stream and preserved in `dump_messages`.
+
+## Tool Approval
+
+!!! note
+    Tool approval requires AI SDK UI v6 or later on the frontend.
+
+Pydantic AI supports human-in-the-loop tool approval workflows with AI SDK UI, allowing users to approve or deny tool executions before they run. See the [deferred tool calls documentation](../deferred-tools.md#human-in-the-loop-tool-approval) for details on setting up tools that require approval.
+
+To enable tool approval streaming, pass `sdk_version=6` to `dispatch_request`:
+
+```py {test="skip" lint="skip"}
+@app.post('/chat')
+async def chat(request: Request) -> Response:
+    return await VercelAIAdapter.dispatch_request(request, agent=agent, sdk_version=6)
+```
+
+When `sdk_version=6`, the adapter will:
+
+1. Emit `tool-approval-request` chunks when tools with `requires_approval=True` are called
+2. Automatically extract approval responses from follow-up requests
+3. Emit `tool-output-denied` chunks for rejected tools
+
+On the frontend, AI SDK UI's [`useChat`](https://ai-sdk.dev/docs/reference/ai-sdk-ui/use-chat) hook handles the approval flow. You can use the [`Confirmation`](https://ai-sdk.dev/elements/components/confirmation) component from AI Elements for a pre-built approval UI, or build your own using the hook's `addToolApprovalResponse` function.

pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@@ -17,7 +17,7 @@
 )
 
 from pydantic import BaseModel, ValidationError
-from typing_extensions import TypeVar
+from typing_extensions import Self, TypeVar
 
 from pydantic_ai import DeferredToolRequests, DeferredToolResults
 from pydantic_ai.agent import AbstractAgent
@@ -127,13 +127,18 @@ class UIAdapter(ABC, Generic[RunInputT, MessageT, EventT, AgentDepsT, OutputData
 
     @classmethod
     async def from_request(
-        cls, request: Request, *, agent: AbstractAgent[AgentDepsT, OutputDataT]
-    ) -> UIAdapter[RunInputT, MessageT, EventT, AgentDepsT, OutputDataT]:
-        """Create an adapter from a request."""
+        cls, request: Request, *, agent: AbstractAgent[AgentDepsT, OutputDataT], **kwargs: Any
+    ) -> Self:
+        """Create an adapter from a request.
+
+        Extra keyword arguments are forwarded to the adapter constructor, allowing subclasses
+        to accept additional adapter-specific parameters.
+        """
         return cls(
             agent=agent,
             run_input=cls.build_run_input(await request.body()),
             accept=request.headers.get('accept'),
+            **kwargs,
         )
 
     @classmethod
@@ -174,6 +179,11 @@ def state(self) -> dict[str, Any] | None:
         """Frontend state from the protocol-specific run input."""
         return None
 
+    @cached_property
+    def deferred_tool_results(self) -> DeferredToolResults | None:
+        """Deferred tool results extracted from the request, used for tool approval workflows."""
+        return None
+
     def transform_stream(
         self,
         stream: AsyncIterator[NativeEvent],
@@ -247,6 +257,9 @@ def run_stream_native(
             output_type = [output_type or self.agent.output_type, DeferredToolRequests]
             toolsets = [*(toolsets or []), toolset]
 
+        if deferred_tool_results is None:
+            deferred_tool_results = self.deferred_tool_results
+
         if isinstance(deps, StateHandler):
             raw_state = self.state or {}
             if isinstance(deps.state, BaseModel):
@@ -356,9 +369,13 @@ async def dispatch_request(
         toolsets: Sequence[AbstractToolset[DispatchDepsT]] | None = None,
         builtin_tools: Sequence[AbstractBuiltinTool] | None = None,
         on_complete: OnCompleteFunc[EventT] | None = None,
+        **kwargs: Any,
     ) -> Response:
         """Handle a protocol-specific HTTP request by running the agent and returning a streaming response of protocol-specific events.
 
+        Extra keyword arguments are forwarded to [`from_request`][pydantic_ai.ui.UIAdapter.from_request],
+        allowing subclasses to accept additional adapter-specific parameters.
+
         Args:
             request: The incoming Starlette/FastAPI request.
             agent: The agent to run.
@@ -379,6 +396,7 @@ async def dispatch_request(
             builtin_tools: Optional additional builtin tools to use for this run.
             on_complete: Optional callback function called when the agent run completes successfully.
                 The callback receives the completed [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] and can optionally yield additional protocol-specific events.
+            **kwargs: Additional keyword arguments forwarded to [`from_request`][pydantic_ai.ui.UIAdapter.from_request].
 
         Returns:
             A streaming Starlette response with protocol-specific events encoded per the request's `Accept` header value.
@@ -395,7 +413,7 @@ async def dispatch_request(
             # The DepsT and OutputDataT come from `agent`, not from `cls`; the cast is necessary to explain this to pyright
             adapter = cast(
                 UIAdapter[RunInputT, MessageT, EventT, DispatchDepsT, DispatchOutputDataT],
-                await cls.from_request(request, agent=cast(AbstractAgent[AgentDepsT, OutputDataT], agent)),
+                await cls.from_request(request, agent=cast(AbstractAgent[AgentDepsT, OutputDataT], agent), **kwargs),
             )
         except ValidationError as e:  # pragma: no cover
             return Response(

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/_adapter.py

@@ -35,10 +35,10 @@
     VideoUrl,
 )
 from ...output import OutputDataT
-from ...tools import AgentDepsT
-from .. import MessagesBuilder, UIAdapter, UIEventStream
+from ...tools import AgentDepsT, DeferredToolResults, ToolDenied
+from .. import MessagesBuilder, UIAdapter
 from ._event_stream import VercelAIEventStream
-from ._utils import dump_provider_metadata, iter_metadata_chunks, load_provider_metadata
+from ._utils import dump_provider_metadata, iter_metadata_chunks, iter_tool_approval_responses, load_provider_metadata
 from .request_types import (
     DataUIPart,
     DynamicToolInputAvailablePart,
@@ -63,8 +63,21 @@
 from .response_types import BaseChunk, DataChunk, FileChunk, SourceDocumentChunk, SourceUrlChunk
 
 if TYPE_CHECKING:
-    pass
-
+    from starlette.requests import Request
+    from starlette.responses import Response
+
+    from ...agent import AbstractAgent
+    from ...agent.abstract import AgentMetadata, Instructions
+    from ...builtin_tools import AbstractBuiltinTool
+    from ...models import KnownModelName, Model
+    from ...output import OutputSpec
+    from ...settings import ModelSettings
+    from ...tools import DeferredToolApprovalResult
+    from ...toolsets import AbstractToolset
+    from ...usage import RunUsage, UsageLimits
+    from .. import UIEventStream
+    from .._adapter import DispatchDepsT, DispatchOutputDataT
+    from .._event_stream import OnCompleteFunc
 
 __all__ = ['VercelAIAdapter']
 
@@ -77,17 +90,92 @@ class VercelAIAdapter(UIAdapter[RequestData, UIMessage, BaseChunk, AgentDepsT, O
 
     _: KW_ONLY
     sdk_version: Literal[5, 6] = 5
-    """Vercel AI SDK version to target. Default is 5 for backwards compatibility."""
+    """Vercel AI SDK version to target. Default is 5 for backwards compatibility.
+
+    Setting `sdk_version=6` enables tool approval streaming for human-in-the-loop workflows.
+    """
 
     @classmethod
     def build_run_input(cls, body: bytes) -> RequestData:
         """Build a Vercel AI run input object from the request body."""
         return request_data_ta.validate_json(body)
 
+    @classmethod
+    async def from_request(
+        cls,
+        request: Request,
+        *,
+        agent: AbstractAgent[AgentDepsT, OutputDataT],
+        sdk_version: Literal[5, 6] = 5,
+        **kwargs: Any,
+    ) -> VercelAIAdapter[AgentDepsT, OutputDataT]:
+        """Extends [`from_request`][pydantic_ai.ui.UIAdapter.from_request] with the `sdk_version` parameter."""
+        return await super().from_request(request, agent=agent, sdk_version=sdk_version, **kwargs)
+
+    @classmethod
+    async def dispatch_request(
+        cls,
+        request: Request,
+        *,
+        agent: AbstractAgent[DispatchDepsT, DispatchOutputDataT],
+        sdk_version: Literal[5, 6] = 5,
+        message_history: Sequence[ModelMessage] | None = None,
+        deferred_tool_results: DeferredToolResults | None = None,
+        model: Model | KnownModelName | str | None = None,
+        instructions: Instructions[DispatchDepsT] = None,
+        deps: DispatchDepsT = None,
+        output_type: OutputSpec[Any] | None = None,
+        model_settings: ModelSettings | None = None,
+        usage_limits: UsageLimits | None = None,
+        usage: RunUsage | None = None,
+        metadata: AgentMetadata[DispatchDepsT] | None = None,
+        infer_name: bool = True,
+        toolsets: Sequence[AbstractToolset[DispatchDepsT]] | None = None,
+        builtin_tools: Sequence[AbstractBuiltinTool] | None = None,
+        on_complete: OnCompleteFunc[BaseChunk] | None = None,
+        **kwargs: Any,
+    ) -> Response:
+        """Extends [`dispatch_request`][pydantic_ai.ui.UIAdapter.dispatch_request] with the `sdk_version` parameter."""
+        return await super().dispatch_request(
+            request,
+            agent=agent,
+            sdk_version=sdk_version,
+            message_history=message_history,
+            deferred_tool_results=deferred_tool_results,
+            model=model,
+            instructions=instructions,
+            deps=deps,
+            output_type=output_type,
+            model_settings=model_settings,
+            usage_limits=usage_limits,
+            usage=usage,
+            metadata=metadata,
+            infer_name=infer_name,
+            toolsets=toolsets,
+            builtin_tools=builtin_tools,
+            on_complete=on_complete,
+            **kwargs,
+        )
+
     def build_event_stream(self) -> UIEventStream[RequestData, BaseChunk, AgentDepsT, OutputDataT]:
         """Build a Vercel AI event stream transformer."""
         return VercelAIEventStream(self.run_input, accept=self.accept, sdk_version=self.sdk_version)
 
+    @cached_property
+    def deferred_tool_results(self) -> DeferredToolResults | None:
+        """Extract deferred tool results from Vercel AI messages with approval responses."""
+        if self.sdk_version < 6:
+            return None
+        approvals: dict[str, bool | DeferredToolApprovalResult] = {}
+        for tool_call_id, approval in iter_tool_approval_responses(self.run_input.messages):
+            if approval.approved:
+                approvals[tool_call_id] = True
+            elif approval.reason:
+                approvals[tool_call_id] = ToolDenied(message=approval.reason)
+            else:
+                approvals[tool_call_id] = False
+        return DeferredToolResults(approvals=approvals) if approvals else None
+
     @cached_property
     def messages(self) -> list[ModelMessage]:
         """Pydantic AI messages from the Vercel AI run input."""

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/_event_stream.py

@@ -4,7 +4,9 @@
 
 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
 
 from pydantic_core import to_json
 
@@ -26,9 +28,9 @@
 )
 from ...output import OutputDataT
 from ...run import AgentRunResultEvent
-from ...tools import AgentDepsT
+from ...tools import AgentDepsT, DeferredToolRequests
 from .. import UIEventStream
-from ._utils import dump_provider_metadata, iter_metadata_chunks
+from ._utils import dump_provider_metadata, iter_metadata_chunks, iter_tool_approval_responses
 from .request_types import RequestData
 from .response_types import (
     BaseChunk,
@@ -46,10 +48,12 @@
     TextDeltaChunk,
     TextEndChunk,
     TextStartChunk,
+    ToolApprovalRequestChunk,
     ToolInputAvailableChunk,
     ToolInputDeltaChunk,
     ToolInputStartChunk,
     ToolOutputAvailableChunk,
+    ToolOutputDeniedChunk,
     ToolOutputErrorChunk,
 )
 
@@ -79,11 +83,20 @@ class VercelAIEventStream(UIEventStream[RequestData, BaseChunk, AgentDepsT, Outp
 
     _: KW_ONLY
     sdk_version: Literal[5, 6] = 5
-    """Vercel AI SDK version to target."""
+    """Vercel AI SDK version to target. Setting to 6 enables tool approval streaming."""
 
     _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
@@ -111,6 +124,16 @@ async def handle_run_result(self, event: AgentRunResultEvent) -> AsyncIterator[B
         pydantic_reason = event.result.response.finish_reason
         if pydantic_reason:
             self._finish_reason = _FINISH_REASON_MAP.get(pydantic_reason, 'other')
+
+        # Emit tool approval requests for deferred approvals (only when sdk_version >= 6)
+        output = event.result.output
+        if self.sdk_version >= 6 and isinstance(output, DeferredToolRequests):
+            for tool_call in output.approvals:
+                yield ToolApprovalRequestChunk(
+                    approval_id=str(uuid4()),
+                    tool_call_id=tool_call.tool_call_id,
+                )
+            return
         return
         yield
 
@@ -247,10 +270,15 @@ async def handle_file(self, part: FilePart) -> AsyncIterator[BaseChunk]:
 
     async def handle_function_tool_result(self, event: FunctionToolResultEvent) -> AsyncIterator[BaseChunk]:
         part = event.result
-        if isinstance(part, RetryPromptPart):
-            yield ToolOutputErrorChunk(tool_call_id=part.tool_call_id, error_text=part.model_response())
+        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:
+            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())
         else:
-            yield ToolOutputAvailableChunk(tool_call_id=part.tool_call_id, output=self._tool_return_output(part))
+            yield ToolOutputAvailableChunk(tool_call_id=tool_call_id, output=self._tool_return_output(part))
 
         # ToolOutputAvailableChunk/ToolOutputErrorChunk.output may hold user parts
         # (e.g. text, images) that Vercel AI does not currently have chunk types for.

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/_utils.py

@@ -4,6 +4,18 @@
 from typing import Any
 
 from pydantic_ai.messages import ProviderDetailsDelta, ToolReturnPart
+from pydantic_ai.ui.vercel_ai.request_types import (
+    DynamicToolInputAvailablePart,
+    DynamicToolInputStreamingPart,
+    DynamicToolOutputAvailablePart,
+    DynamicToolOutputErrorPart,
+    ToolApprovalResponded,
+    ToolInputAvailablePart,
+    ToolInputStreamingPart,
+    ToolOutputAvailablePart,
+    ToolOutputErrorPart,
+    UIMessage,
+)
 from pydantic_ai.ui.vercel_ai.response_types import (
     DataChunk,
     FileChunk,
@@ -81,3 +93,26 @@ def iter_metadata_chunks(
         for item in possible:  # type: ignore[reportUnknownMemberType]
             if isinstance(item, _DATA_CHUNK_TYPES):  # pragma: no branch
                 yield item
+
+
+_TOOL_PART_TYPES = (
+    ToolInputStreamingPart,
+    ToolInputAvailablePart,
+    ToolOutputAvailablePart,
+    ToolOutputErrorPart,
+    DynamicToolInputStreamingPart,
+    DynamicToolInputAvailablePart,
+    DynamicToolOutputAvailablePart,
+    DynamicToolOutputErrorPart,
+)
+
+
+def iter_tool_approval_responses(
+    messages: list[UIMessage],
+) -> Iterator[tuple[str, ToolApprovalResponded]]:
+    """Yield `(tool_call_id, approval)` for each responded tool approval in assistant messages."""
+    for msg in messages:
+        if msg.role == 'assistant':
+            for part in msg.parts:
+                if isinstance(part, _TOOL_PART_TYPES) and isinstance(part.approval, ToolApprovalResponded):
+                    yield part.tool_call_id, part.approval