Wire tool approval into AI SDK UI protocol

Author: anbuzin

src/vercel_ai_sdk/__init__.py

@@ -1,7 +1,7 @@
 from . import ai_gateway, ai_sdk_ui, anthropic, mcp, openai
 from .core import telemetry
 from .core.checkpoint import Checkpoint
-from .core.hooks import Hook, hook
+from .core.hooks import Hook, ToolApproval, hook
 from .core.llm import LanguageModel
 
 # Re-export core types
@@ -51,6 +51,7 @@
     "StreamResult",
     "Hook",
     "HookPart",
+    "ToolApproval",
     "StructuredOutputPart",
     "Checkpoint",
     # Functions

src/vercel_ai_sdk/agent/__init__.py

@@ -1,4 +1,4 @@
 from . import local, proto, tools, vercel
-from .agent import Agent, ToolApproval
+from .agent import Agent
 
-__all__ = ["Agent", "ToolApproval", "proto", "tools", "local", "vercel"]
+__all__ = ["Agent", "proto", "tools", "local", "vercel"]

src/vercel_ai_sdk/agent/agent.py

@@ -4,20 +4,12 @@
 import dataclasses
 from typing import Any
 
-import pydantic
-
 import vercel_ai_sdk as ai
 
 from . import proto
 from .tools import BUILTIN_TOOLS, _filesystem
 
 
-@ai.hook
-class ToolApproval(pydantic.BaseModel):
-    granted: bool
-    reason: str | None = None
-
-
 @dataclasses.dataclass
 class Agent:
     """
@@ -48,7 +40,7 @@ async def _execute_tool(
         """
         # TODO: mypy doesn't support class decorators that change the class type —
         # @ai.hook returns type[Hook[T]] but mypy still sees the original BaseModel.
-        approval = await ToolApproval.create(  # type: ignore[attr-defined]
+        approval = await ai.ToolApproval.create(  # type: ignore[attr-defined]
             f"approve_{tc.tool_call_id}",
             metadata={"tool_name": tc.tool_name, "tool_args": tc.tool_args},
         )

src/vercel_ai_sdk/ai_sdk_ui/adapter.py

@@ -11,6 +11,7 @@
 from typing import Any, Literal
 
 from .. import core
+from ..core import hooks
 from . import protocol, ui_message
 
 # ============================================================================
@@ -69,6 +70,7 @@ def __init__(self) -> None:
         self.started_tool_calls: set[str] = set()
         self.emitted_tool_results: set[str] = set()
         self.pending_tool_calls: set[str] = set()
+        self.emitted_approval_requests: set[str] = set()
 
     def close_open_blocks(self) -> list[protocol.UIMessageStreamPart]:
         """Close any open reasoning/text blocks, returning parts to emit."""
@@ -94,6 +96,7 @@ def reset_tool_tracking(self) -> None:
         self.started_tool_calls = set()
         self.emitted_tool_results = set()
         self.pending_tool_calls = set()
+        self.emitted_approval_requests = set()
 
     def begin_message(
         self, msg: core.messages.Message
@@ -130,6 +133,22 @@ def begin_message(
         return parts
 
 
+def _tool_call_id_from_approval_hook(
+    hook_part: core.messages.HookPart,
+) -> str | None:
+    """Extract tool_call_id from a ToolApproval HookPart.
+
+    Returns the tool_call_id if this is a ToolApproval hook whose hook_id
+    follows the ``approve_{tool_call_id}`` convention, otherwise None.
+    """
+    if hook_part.hook_type != hooks.ToolApproval.hook_type:  # type: ignore[attr-defined]
+        return None
+    prefix = "approve_"
+    if hook_part.hook_id.startswith(prefix):
+        return hook_part.hook_id[len(prefix) :]
+    return None
+
+
 async def to_ui_message_stream(
     messages: AsyncIterable[core.messages.Message],
 ) -> AsyncGenerator[protocol.UIMessageStreamPart]:
@@ -256,6 +275,33 @@ async def to_ui_message_stream(
                                 output=result,
                             )
 
+            # Pass 3: Hook-based tool approvals
+            for msg_part in msg.parts:
+                if not isinstance(msg_part, core.messages.HookPart):
+                    continue
+                approval_tc_id = _tool_call_id_from_approval_hook(msg_part)
+                if approval_tc_id is None:
+                    continue
+
+                if msg_part.status == "pending":
+                    if approval_tc_id not in state.emitted_approval_requests:
+                        state.emitted_approval_requests.add(approval_tc_id)
+                        yield protocol.ToolApprovalRequestPart(
+                            approval_id=msg_part.hook_id,
+                            tool_call_id=approval_tc_id,
+                        )
+                elif msg_part.status == "resolved":
+                    resolution = msg_part.resolution or {}
+                    if not resolution.get("granted", False):
+                        yield protocol.ToolOutputDeniedPart(
+                            tool_call_id=approval_tc_id,
+                        )
+                elif msg_part.status == "cancelled":
+                    yield protocol.ToolOutputErrorPart(
+                        tool_call_id=approval_tc_id,
+                        error_text="Hook cancelled",
+                    )
+
     # Final cleanup
     for part in state.finish_step():
         yield part
@@ -333,6 +379,10 @@ def to_messages(
 ) -> list[core.messages.Message]:
     """Convert AI SDK v6 UI messages to internal Message format.
 
+    As a side-effect, tool parts in ``approval-responded`` state trigger
+    ``ToolApproval.resolve()`` so the agent loop can resume execution
+    without the caller needing to handle approval routing explicitly.
+
     Args:
         ui_messages: List of UIMessage objects from the AI SDK v6 frontend.
 
@@ -375,6 +425,20 @@ def to_messages(
                             result=_normalize_tool_result(tp.output),
                         )
                     )
+                    # Side-effect: resolve ToolApproval hooks from approval
+                    # responses so the agent loop can resume execution.
+                    if (
+                        tp.state == "approval-responded"
+                        and tp.approval is not None
+                        and tp.approval.approved is not None
+                    ):
+                        hooks.ToolApproval.resolve(  # type: ignore[attr-defined]
+                            tp.approval.id,
+                            {
+                                "granted": tp.approval.approved,
+                                "reason": tp.approval.reason,
+                            },
+                        )
 
                 case (
                     ui_message.UIStepStartPart()

src/vercel_ai_sdk/ai_sdk_ui/ui_message.py

@@ -37,11 +37,11 @@ class UIReasoningPart(pydantic.BaseModel):
 # Tool invocation states in AI SDK v6:
 # - "input-streaming": Tool arguments are being streamed
 # - "input-available": Tool arguments are complete, ready for execution
-# - "approval-requested": Tool requires user approval (TODO: approval workflow)
-# - "approval-responded": User has responded to approval (TODO: approval workflow)
+# - "approval-requested": Tool requires user approval before execution
+# - "approval-responded": User has responded to approval request
 # - "output-available": Tool has been executed, result is available
 # - "output-error": Tool execution failed
-# - "output-denied": Tool execution was denied by user (TODO: approval workflow)
+# - "output-denied": Tool execution was denied by user
 UIToolInvocationState = Literal[
     "input-streaming",
     "input-available",
@@ -78,6 +78,21 @@ class UIStepStartPart(pydantic.BaseModel):
     type: Literal["step-start"]
 
 
+class UIToolApproval(pydantic.BaseModel):
+    """Approval state on a tool part (AI SDK v6 protocol).
+
+    Present when a tool requires user approval before execution.
+    ``id`` matches the hook label used by the ToolApproval hook.
+    ``approved`` is None while awaiting a response, True/False after.
+    """
+
+    model_config = pydantic.ConfigDict(populate_by_name=True)
+
+    id: str
+    approved: bool | None = None
+    reason: str | None = None
+
+
 class UIToolPart(pydantic.BaseModel):
     """Tool part with dynamic type pattern: tool-{toolName}.
 
@@ -95,8 +110,7 @@ class UIToolPart(pydantic.BaseModel):
     input: str | dict[str, Any] | None = None  # JSON string or parsed dict
     output: Any | None = None
     error_text: str | None = pydantic.Field(default=None, alias="errorText")
-    # TODO: title, providerExecuted, preliminary fields
-    # TODO: approval workflow (approval object)
+    approval: UIToolApproval | None = None
 
     @property
     def tool_name(self) -> str:

src/vercel_ai_sdk/core/hooks.py

@@ -72,7 +72,7 @@ class Hook[T: pydantic.BaseModel]:
     """
 
     _schema: ClassVar[type[pydantic.BaseModel]]
-    _hook_type: ClassVar[str]
+    hook_type: ClassVar[str]
 
     @classmethod
     async def create(cls, label: str, metadata: dict[str, Any] | None = None) -> T:
@@ -113,7 +113,7 @@ async def create(cls, label: str, metadata: dict[str, Any] | None = None) -> T:
         future: asyncio.Future[dict[str, Any]] = asyncio.Future()
         suspension = rt_mod.HookSuspension(
             label=label,
-            hook_type=cls._hook_type,
+            hook_type=cls.hook_type,
             metadata=metadata or {},
             future=future,
         )
@@ -142,7 +142,7 @@ async def create(cls, label: str, metadata: dict[str, Any] | None = None) -> T:
                 parts=[
                     messages_.HookPart(
                         hook_id=label,
-                        hook_type=cls._hook_type,
+                        hook_type=cls.hook_type,
                         status="resolved",
                         metadata=hook_metadata,
                         resolution=resolution,
@@ -215,7 +215,7 @@ async def cancel(cls, label: str, reason: str | None = None) -> None:
                 parts=[
                     messages_.HookPart(
                         hook_id=label,
-                        hook_type=cls._hook_type,
+                        hook_type=cls.hook_type,
                         status="cancelled",
                         metadata=hook_metadata,
                     )
@@ -235,9 +235,22 @@ def hook[T: pydantic.BaseModel](cls: type[T]) -> type[Hook[T]]:
         (Hook,),
         {
             "_schema": cls,
-            "_hook_type": cls.__name__,
+            "hook_type": cls.__name__,
             "__doc__": cls.__doc__,
         },
     )
 
     return hook_impl
+
+
+@hook
+class ToolApproval(pydantic.BaseModel):
+    """Prewired hook for tool call approval.
+
+    Used by the AI SDK UI adapter to bridge the protocol's
+    tool-approval-request / approval-responded flow to the
+    hook system.
+    """
+
+    granted: bool
+    reason: str | None = None

tests/ai_sdk_ui/test_adapter.py

@@ -8,7 +8,7 @@
 
 import vercel_ai_sdk as ai
 from vercel_ai_sdk.ai_sdk_ui import adapter, ui_message
-from vercel_ai_sdk.core import messages
+from vercel_ai_sdk.core import hooks, messages
 
 from ..conftest import MockLLM
 
@@ -492,3 +492,88 @@ def test_ui_skips_unsupported_parts() -> None:
 
     internal = adapter.to_messages([ui_msg])
     assert len(internal[0].parts) == 2
+
+
+# -----------------------------------------------------------------------------
+# Tool approval (human-in-the-loop) tests
+# -----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_tool_approval_hook_emits_approval_request() -> None:
+    """Pending ToolApproval HookPart emits tool-approval-request on the wire."""
+    msgs = [
+        # Tool pending (args complete, awaiting approval)
+        messages.Message(
+            id="msg-1",
+            role="assistant",
+            parts=[
+                messages.ToolPart(
+                    tool_call_id="tc-1",
+                    tool_name="rm_rf",
+                    tool_args='{"path": "/"}',
+                    status="pending",
+                    state="done",
+                ),
+            ],
+        ),
+        # Hook pending (approval requested)
+        messages.Message(
+            id="msg-1",
+            role="assistant",
+            parts=[
+                messages.HookPart(
+                    hook_id="approve_tc-1",
+                    hook_type=hooks.ToolApproval.hook_type,  # type: ignore[attr-defined]
+                    status="pending",
+                    metadata={"tool_name": "rm_rf", "tool_args": '{"path": "/"}'},
+                ),
+            ],
+        ),
+    ]
+
+    event_types = await get_event_types(msgs)
+    assert event_types == [
+        "start",
+        "start-step",
+        "tool-input-start",
+        "tool-input-available",
+        "tool-approval-request",
+        "finish-step",
+        "finish",
+    ]
+
+
+def test_approval_responded_resolves_hook() -> None:
+    """to_messages() resolves the ToolApproval hook for approval-responded parts."""
+    label = "approve_tc-42"
+    raw_messages = [
+        {
+            "id": "msg-1",
+            "role": "assistant",
+            "parts": [
+                {
+                    "type": "tool-dangerous_action",
+                    "toolCallId": "tc-42",
+                    "state": "approval-responded",
+                    "input": '{"x": 1}',
+                    "approval": {
+                        "id": label,
+                        "approved": True,
+                        "reason": "looks safe",
+                    },
+                }
+            ],
+        },
+    ]
+
+    # Clean up any leftover state from other tests
+    hooks._pending_resolutions.pop(label, None)
+
+    ui_msgs = [ui_message.UIMessage.model_validate(m) for m in raw_messages]
+    adapter.to_messages(ui_msgs)
+
+    # The side-effect should have pre-registered the resolution
+    assert label in hooks._pending_resolutions
+    resolution = hooks._pending_resolutions.pop(label)
+    assert resolution == {"granted": True, "reason": "looks safe"}