Add tool approval integration for Vercel AI adapter

Author: bendrucker

docs/ui/vercel-ai.md

@@ -2,6 +2,9 @@
 
 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.
 
+!!! note "AI SDK Version Compatibility"
+    The base protocol is compatible with AI SDK v5 and later. However, [tool approval](#tool-approval) (human-in-the-loop) features require **AI SDK v6 or later**.
+
 ## Usage
 
 The [`VercelAIAdapter`][pydantic_ai.ui.vercel_ai.VercelAIAdapter] class is responsible for transforming agent run input received from the frontend into arguments for [`Agent.run_stream_events()`](../agents.md#running-agents), running the agent, and then transforming Pydantic AI events into Vercel AI events. The event stream transformation is handled by the [`VercelAIEventStream`][pydantic_ai.ui.vercel_ai.VercelAIEventStream] class, but you typically won't use this directly.
@@ -81,3 +84,58 @@ async def chat(request: Request) -> Response:
     sse_event_stream = adapter.encode_stream(event_stream)
     return StreamingResponse(sse_event_stream, media_type=accept)
 ```
+
+## Tool Approval
+
+Pydantic AI supports [AI SDK's human-in-the-loop tool approval](https://ai-sdk.dev/cookbook/next/human-in-the-loop) workflow, allowing users to approve or deny tool executions before they run.
+
+!!! warning "Requires AI SDK v6"
+    Tool approval is an AI SDK v6 feature. The `tool-approval-request` and `tool-output-denied` stream chunks, along with the `approval` field on tool parts, are not available in AI SDK v5.
+
+### How It Works
+
+1. **Tool requests approval**: When an agent calls a tool with `requires_approval=True`, Pydantic AI emits a `tool-approval-request` chunk instead of executing the tool immediately.
+
+2. **User decides**: The AI SDK frontend displays an approval UI. The user can approve or deny the tool execution.
+
+3. **Response is sent**: The frontend sends the approval decision back to the server in a follow-up request.
+
+4. **Tool executes or is denied**: If approved, the tool runs normally. If denied, Pydantic AI emits a `tool-output-denied` chunk and informs the model that the tool was rejected.
+
+### Server-Side Setup
+
+To enable tool approval, define your tool with `requires_approval=True` and include [`DeferredToolRequests`][pydantic_ai.tools.DeferredToolRequests] in your agent's output types:
+
+```py
+from pydantic_ai import Agent
+from pydantic_ai.tools import DeferredToolRequests
+
+agent: Agent[None, str | DeferredToolRequests] = Agent(
+    'openai:gpt-5',
+    output_type=[str, DeferredToolRequests],
+)
+
+@agent.tool_plain(requires_approval=True)
+def delete_file(path: str) -> str:
+    """Delete a file from the filesystem."""
+    # This won't execute until the user approves
+    os.remove(path)
+    return f'Deleted {path}'
+```
+
+When processing a follow-up request with approval responses, extract and pass the deferred tool results:
+
+```py
+@app.post('/chat')
+async def chat(request: Request) -> Response:
+    adapter = await VercelAIAdapter.from_request(request, agent=agent)
+    return adapter.streaming_response(
+        adapter.run_stream(deferred_tool_results=adapter.deferred_tool_results)
+    )
+```
+
+### Client-Side Setup
+
+On the frontend, use AI SDK v6's [`useChat`](https://v6.ai-sdk.dev/docs/ai-sdk-ui/use-chat) hook with the [`Confirmation`](https://ai-sdk.dev/elements/components/confirmation) component or the `addToolApprovalResponse` function to handle approval UI.
+
+See the [AI SDK Human-in-the-Loop Cookbook](https://ai-sdk.dev/cookbook/next/human-in-the-loop) for complete frontend examples.

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/_adapter.py

@@ -35,7 +35,7 @@
     VideoUrl,
 )
 from ...output import OutputDataT
-from ...tools import AgentDepsT
+from ...tools import AgentDepsT, DeferredToolResults, ToolApproved, ToolDenied
 from .. import MessagesBuilder, UIAdapter, UIEventStream
 from ._event_stream import VercelAIEventStream
 from .request_types import (
@@ -51,6 +51,7 @@
     SourceUrlUIPart,
     StepStartUIPart,
     TextUIPart,
+    ToolApprovalResponded,
     ToolInputAvailablePart,
     ToolOutputAvailablePart,
     ToolOutputErrorPart,
@@ -87,6 +88,54 @@ def messages(self) -> list[ModelMessage]:
         """Pydantic AI messages from the Vercel AI run input."""
         return self.load_messages(self.run_input.messages)
 
+    @cached_property
+    def deferred_tool_results(self) -> DeferredToolResults | None:
+        """Extract deferred tool results from tool parts with approval responses.
+
+        When the Vercel AI SDK client responds to a tool-approval-request, it sends
+        the approval decision in the tool part's `approval` field. This method extracts
+        those responses and converts them to Pydantic AI's `DeferredToolResults` format.
+
+        Returns:
+            DeferredToolResults if any tool parts have approval responses, None otherwise.
+        """
+        return self.extract_deferred_tool_results(self.run_input.messages)
+
+    @classmethod
+    def extract_deferred_tool_results(cls, messages: Sequence[UIMessage]) -> DeferredToolResults | None:
+        """Extract deferred tool results from UI messages.
+
+        Args:
+            messages: The UI messages to scan for approval responses.
+
+        Returns:
+            DeferredToolResults if any tool parts have approval responses, None otherwise.
+        """
+        approvals: dict[str, bool | ToolApproved | ToolDenied] = {}
+
+        for msg in messages:
+            if msg.role != 'assistant':
+                continue
+
+            for part in msg.parts:
+                if not isinstance(part, ToolUIPart | DynamicToolUIPart):
+                    continue
+
+                approval = part.approval
+                if approval is None or not isinstance(approval, ToolApprovalResponded):
+                    continue
+
+                tool_call_id = part.tool_call_id
+                if approval.approved:
+                    approvals[tool_call_id] = ToolApproved()
+                else:
+                    approvals[tool_call_id] = ToolDenied(message=approval.reason or 'The tool call was denied.')
+
+        if not approvals:
+            return None
+
+        return DeferredToolResults(approvals=approvals)
+
     @classmethod
     def load_messages(cls, messages: Sequence[UIMessage]) -> list[ModelMessage]:  # noqa: C901
         """Transform Vercel AI messages into Pydantic AI messages."""

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/_event_stream.py

@@ -5,6 +5,7 @@
 from collections.abc import AsyncIterator, Mapping
 from dataclasses import dataclass
 from typing import Any
+from uuid import uuid4
 
 from pydantic_core import to_json
 
@@ -25,9 +26,15 @@
 )
 from ...output import OutputDataT
 from ...run import AgentRunResultEvent
-from ...tools import AgentDepsT
+from ...tools import AgentDepsT, DeferredToolRequests
 from .. import UIEventStream
-from .request_types import RequestData
+from .request_types import (
+    DynamicToolUIPart,
+    RequestData,
+    ToolApprovalResponded,
+    ToolUIPart,
+    UIMessage,
+)
 from .response_types import (
     BaseChunk,
     DoneChunk,
@@ -44,10 +51,12 @@
     TextDeltaChunk,
     TextEndChunk,
     TextStartChunk,
+    ToolApprovalRequestChunk,
     ToolInputAvailableChunk,
     ToolInputDeltaChunk,
     ToolInputStartChunk,
     ToolOutputAvailableChunk,
+    ToolOutputDeniedChunk,
     ToolOutputErrorChunk,
 )
 
@@ -71,12 +80,35 @@ def _json_dumps(obj: Any) -> str:
     return to_json(obj).decode('utf-8')
 
 
+def _extract_denied_tool_ids(messages: list[UIMessage]) -> set[str]:
+    """Extract tool_call_ids that were denied from UI messages."""
+    denied_ids: set[str] = set()
+    for msg in messages:
+        if msg.role != 'assistant':
+            continue
+        for part in msg.parts:
+            if not isinstance(part, ToolUIPart | DynamicToolUIPart):
+                continue
+            approval = part.approval
+            if isinstance(approval, ToolApprovalResponded) and not approval.approved:
+                denied_ids.add(part.tool_call_id)
+    return denied_ids
+
+
 @dataclass
 class VercelAIEventStream(UIEventStream[RequestData, BaseChunk, AgentDepsT, OutputDataT]):
     """UI event stream transformer for the Vercel AI protocol."""
 
     _step_started: bool = False
     _finish_reason: FinishReason = None
+    _denied_tool_ids: set[str] | None = None
+
+    @property
+    def denied_tool_ids(self) -> set[str]:
+        """Get the set of tool_call_ids that were denied by the user."""
+        if self._denied_tool_ids is None:
+            self._denied_tool_ids = _extract_denied_tool_ids(self.run_input.messages)
+        return self._denied_tool_ids
 
     @property
     def response_headers(self) -> Mapping[str, str] | None:
@@ -105,8 +137,15 @@ 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)
-        return
-        yield
+
+        # Emit tool approval requests for deferred approvals
+        output = event.result.output
+        if isinstance(output, DeferredToolRequests):
+            for tool_call in output.approvals:
+                yield ToolApprovalRequestChunk(
+                    approval_id=str(uuid4()),
+                    tool_call_id=tool_call.tool_call_id,
+                )
 
     async def on_error(self, error: Exception) -> AsyncIterator[BaseChunk]:
         self._finish_reason = 'error'
@@ -203,10 +242,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
+        if 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))
 
         # ToolCallResultEvent.content may hold user parts (e.g. text, images) that Vercel AI does not currently have events for
 

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

@@ -1,7 +1,9 @@
 """Vercel AI request types (UI messages).
 
 Converted to Python from:
-https://github.com/vercel/ai/blob/ai%405.0.59/packages/ai/src/ui/ui-messages.ts
+https://github.com/vercel/ai/blob/ai%406.0.0-beta.159/packages/ai/src/ui/ui-messages.ts
+
+Tool approval types (`ToolApprovalRequested`, `ToolApprovalResponded`) require AI SDK v6 or later.
 """
 
 from abc import ABC
@@ -110,6 +112,30 @@ class DataUIPart(BaseUIPart):
     data: Any
 
 
+class ToolApprovalRequested(CamelBaseModel):
+    """Tool approval in requested state (awaiting user response)."""
+
+    id: str
+    """The approval request ID."""
+
+
+class ToolApprovalResponded(CamelBaseModel):
+    """Tool approval in responded state (user has approved or denied)."""
+
+    id: str
+    """The approval request ID."""
+
+    approved: bool
+    """Whether the user approved the tool call."""
+
+    reason: str | None = None
+    """Optional reason for the approval or denial."""
+
+
+ToolApproval = ToolApprovalRequested | ToolApprovalResponded
+"""Union of tool approval states."""
+
+
 # Tool part states as separate models
 class ToolInputStreamingPart(BaseUIPart):
     """Tool part in input-streaming state."""
@@ -119,6 +145,7 @@ class ToolInputStreamingPart(BaseUIPart):
     state: Literal['input-streaming'] = 'input-streaming'
     input: Any | None = None
     provider_executed: bool | None = None
+    approval: ToolApproval | None = None
 
 
 class ToolInputAvailablePart(BaseUIPart):
@@ -130,6 +157,7 @@ class ToolInputAvailablePart(BaseUIPart):
     input: Any | None = None
     provider_executed: bool | None = None
     call_provider_metadata: ProviderMetadata | None = None
+    approval: ToolApproval | None = None
 
 
 class ToolOutputAvailablePart(BaseUIPart):
@@ -143,6 +171,7 @@ class ToolOutputAvailablePart(BaseUIPart):
     provider_executed: bool | None = None
     call_provider_metadata: ProviderMetadata | None = None
     preliminary: bool | None = None
+    approval: ToolApproval | None = None
 
 
 class ToolOutputErrorPart(BaseUIPart):
@@ -156,6 +185,7 @@ class ToolOutputErrorPart(BaseUIPart):
     error_text: str
     provider_executed: bool | None = None
     call_provider_metadata: ProviderMetadata | None = None
+    approval: ToolApproval | None = None
 
 
 ToolUIPart = ToolInputStreamingPart | ToolInputAvailablePart | ToolOutputAvailablePart | ToolOutputErrorPart
@@ -171,6 +201,7 @@ class DynamicToolInputStreamingPart(BaseUIPart):
     tool_call_id: str
     state: Literal['input-streaming'] = 'input-streaming'
     input: Any | None = None
+    approval: ToolApproval | None = None
 
 
 class DynamicToolInputAvailablePart(BaseUIPart):
@@ -182,6 +213,7 @@ class DynamicToolInputAvailablePart(BaseUIPart):
     state: Literal['input-available'] = 'input-available'
     input: Any
     call_provider_metadata: ProviderMetadata | None = None
+    approval: ToolApproval | None = None
 
 
 class DynamicToolOutputAvailablePart(BaseUIPart):
@@ -195,6 +227,7 @@ class DynamicToolOutputAvailablePart(BaseUIPart):
     output: Any
     call_provider_metadata: ProviderMetadata | None = None
     preliminary: bool | None = None
+    approval: ToolApproval | None = None
 
 
 class DynamicToolOutputErrorPart(BaseUIPart):
@@ -207,6 +240,7 @@ class DynamicToolOutputErrorPart(BaseUIPart):
     input: Any
     error_text: str
     call_provider_metadata: ProviderMetadata | None = None
+    approval: ToolApproval | None = None
 
 
 DynamicToolUIPart = (

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/response_types.py

@@ -1,7 +1,9 @@
 """Vercel AI response types (SSE chunks).
 
 Converted to Python from:
-https://github.com/vercel/ai/blob/ai%405.0.59/packages/ai/src/ui-message-stream/ui-message-chunks.ts
+https://github.com/vercel/ai/blob/ai%406.0.0-beta.159/packages/ai/src/ui-message-stream/ui-message-chunks.ts
+
+Tool approval types (`ToolApprovalRequestChunk`, `ToolOutputDeniedChunk`) require AI SDK v6 or later.
 """
 
 from abc import ABC