Add require_approval to @tool

examples/fastapi-vite/backend/agent.py

@@ -1,11 +1,11 @@
 """Agent logic for the chat demo.
 
 Demonstrates human-in-the-loop tool approval using ToolApproval hooks.
-Every tool call is gated behind user approval before execution.
+``talk_to_mothership`` is gated behind user approval via the
+``require_approval=True`` flag on the tool.
 """
 
 import asyncio
-from collections.abc import AsyncGenerator
 
 import ai
 
@@ -42,7 +42,7 @@ async def get_population(city: str) -> int:
     return {"new york": 8_336_817, "tokyo": 13_960_000}.get(city.lower(), 1_000_000)
 
 
-@ai.tool
+@ai.tool(require_approval=True)
 async def talk_to_mothership(question: str) -> ai.SubAgentTool:
     """Contact the mothership for important decisions."""
     mothership = ai.agent()
@@ -58,65 +58,4 @@ async def talk_to_mothership(question: str) -> ai.SubAgentTool:
 TOOLS: list[ai.AgentTool] = [get_weather, get_population, talk_to_mothership]
 
 
-class ChatAgent(ai.Agent):
-    """Agent graph with human-in-the-loop tool approval.
-
-    Loops: stream LLM -> request approval -> execute tools -> repeat.
-    The ToolApproval hook suspends execution and emits an approval-
-    request event on the SSE stream.  The frontend displays Approve /
-    Reject buttons and sends the decision back on the next request.
-    """
-
-    async def loop(self, context: ai.Context) -> AsyncGenerator[ai.events.AgentEvent]:
-        while context.keep_running():
-            async with (
-                ai.stream(context=context) as s,
-                ai.ToolRunner() as tr,
-            ):
-                async for event in ai.util.merge(s, tr.events()):
-                    yield event
-                    if isinstance(event, ai.events.ToolEnd):
-                        tc = _resolve(context, event.tool_call)
-                        tr.schedule(tc)
-
-                context.add(s.message)
-                context.add(tr.get_tool_message())
-
-
-chat_agent = ChatAgent(tools=TOOLS)
-
-
-def _resolve(
-    context: ai.Context, tool_call: ai.messages.ToolCallPart
-) -> ai.ToolCallLike:
-    tc = context.resolve(tool_call)
-    if tc.name == "talk_to_mothership":
-        return lambda: _execute_with_approval(tc)
-    else:
-        return tc
-
-
-async def _execute_with_approval(tc: ai.ToolCall) -> ai.events.ToolCallResult:
-    """Execute a tool call only after the user grants approval.
-
-    Creates a ToolApproval hook that suspends execution until the
-    frontend responds with an approve/reject decision.
-    """
-    try:
-        approval = await ai.hook(
-            f"approve_{tc.id}",
-            payload=ai.tools.ToolApproval,
-            metadata={"tool_name": tc.name, "tool_kwargs": tc.kwargs},
-        )
-    except ai.agents.hooks.HookPendingError as e:
-        return ai.pending_tool_result(e.hook, tool_call_id=tc.id, tool_name=tc.name)
-
-    if approval.granted:
-        return await tc()
-
-    return ai.tool_result(
-        tool_call_id=tc.id,
-        tool_name=tc.name,
-        result="Tool call was denied by the user.",
-        is_error=True,
-    )
+chat_agent = ai.Agent(tools=TOOLS)

examples/multiagent-textual/server.py

@@ -78,7 +78,7 @@ class GatedCall:
     path before the underlying tool runs.
     """
 
-    def __init__(self, tc: ai.ToolCall, label: str) -> None:
+    def __init__(self, tc: ai.ToolCallLike, label: str) -> None:
         self._tc = tc
         self._label = label
 

examples/samples/agent_hooks.py

@@ -1,81 +1,27 @@
 """Human-in-the-loop approval hooks.
 
 Demonstrates the function-based hook API:
-  - await hook("label", payload=Model) to suspend inside the loop
+  - mark a tool with ``require_approval=True`` to gate its execution
+    behind an approval hook
   - resolve_hook("label", data) to unblock from outside
   - Hook signals arrive as HookEvent events
-
-The custom loop uses the concurrent ``ToolRunner`` flow: tools are
-scheduled and run concurrently as the model emits them.  The approval
-hook is awaited inside a ``ToolCall``-shaped wrapper that is scheduled
-in place of the bare tool call, so gating composes naturally with the
-runner's merge-and-iterate behaviour.
 """
 
 import asyncio
-from collections.abc import AsyncGenerator
 
 import ai
 
 
-@ai.tool
+@ai.tool(require_approval=True)
 async def contact_mothership(query: str) -> str:
     """Contact the mothership for important decisions."""
     return "Soon."
 
 
-class GatedCall:
-    """ToolCall-shaped wrapper that awaits an approval hook before executing.
-
-    ``ToolRunner.schedule`` only consumes the ``__call__`` shape of
-    ``ToolCall``; this wrapper supplies the same shape while inserting
-    the hook await + denial path before the underlying tool runs.
-    """
-
-    def __init__(self, tc: ai.ToolCall) -> None:
-        self._tc = tc
-
-    async def __call__(self) -> ai.events.ToolCallResult:
-        tc = self._tc
-        approval = await ai.hook(
-            f"approve_{tc.id}",
-            payload=ai.tools.ToolApproval,
-            metadata={"tool": tc.name, "kwargs": tc.kwargs},
-        )
-        if approval.granted:
-            return await tc()
-        return ai.tool_result(
-            tool_call_id=tc.id,
-            tool_name=tc.name,
-            result=f"Rejected: {approval.reason}",
-            is_error=True,
-        )
-
-
-class ApprovalAgent(ai.Agent):
-    async def loop(self, context: ai.Context) -> AsyncGenerator[ai.events.AgentEvent]:
-        while context.keep_running():
-            async with (
-                ai.stream(context=context) as s,
-                ai.ToolRunner() as tr,
-            ):
-                async for event in ai.util.merge(s, tr.events()):
-                    yield event
-                    if isinstance(event, ai.events.ToolEnd):
-                        tc = context.resolve(event.tool_call)
-                        if tc.name == "contact_mothership":
-                            tr.schedule(GatedCall(tc))
-                        else:
-                            tr.schedule(tc)
-
-                context.add(s.message)
-                context.add(tr.get_tool_message())
-
-
 async def main() -> None:
     model = ai.ai_gateway("anthropic/claude-sonnet-4")
 
-    my_agent = ApprovalAgent(tools=[contact_mothership])
+    my_agent = ai.Agent(tools=[contact_mothership])
 
     messages = [
         ai.system_message(

examples/samples/agent_hooks_serverless.py

@@ -13,14 +13,13 @@
 """
 
 import asyncio
-from collections.abc import AsyncGenerator
 
 import ai
 
 FILES_DELETED = set()
 
 
-@ai.tool
+@ai.tool(require_approval=True)
 async def delete_file(path: str) -> str:
     """Delete a file at the given path."""
     print("FILE DELETED:", path)
@@ -39,61 +38,10 @@ async def audit_log(message: str) -> str:
     return f"Logged {message!r}"
 
 
-class GatedCall:
-    """ToolCall-shaped wrapper that awaits an approval hook before executing.
-
-    ``ToolRunner.schedule`` only consumes the ``__call__`` shape of
-    ``ToolCall``; this wrapper supplies the same shape while inserting
-    the hook await + denial path before the underlying tool runs.
-    """
-
-    def __init__(self, tc: ai.ToolCall) -> None:
-        self._tc = tc
-
-    async def __call__(self) -> ai.events.ToolCallResult:
-        tc = self._tc
-        try:
-            approval = await ai.hook(
-                f"approve_{tc.id}",
-                payload=ai.tools.ToolApproval,
-                metadata={"tool": tc.name, "kwargs": tc.kwargs},
-            )
-        except ai.agents.hooks.HookPendingError as e:
-            return ai.pending_tool_result(e.hook, tool_call_id=tc.id, tool_name=tc.name)
-        if approval.granted:
-            return await tc()
-        return ai.tool_result(
-            tool_call_id=tc.id,
-            tool_name=tc.name,
-            result=f"Rejected: {approval.reason}",
-            is_error=True,
-        )
-
-
-class ConfirmAgent(ai.Agent):
-    async def loop(self, context: ai.Context) -> AsyncGenerator[ai.events.AgentEvent]:
-        while context.keep_running():
-            async with (
-                ai.stream(context=context) as s,
-                ai.ToolRunner() as tr,
-            ):
-                async for event in ai.util.merge(s, tr.events()):
-                    yield event
-                    if isinstance(event, ai.events.ToolEnd):
-                        tc = context.resolve(event.tool_call)
-                        if tc.name == "delete_file":
-                            tr.schedule(GatedCall(tc))
-                        else:
-                            tr.schedule(tc)
-
-                context.add(s.message)
-                context.add(tr.get_tool_message())
-
-
 async def main() -> None:
     model = ai.ai_gateway("anthropic/claude-sonnet-4")
 
-    my_agent = ConfirmAgent(tools=[delete_file, audit_log])
+    my_agent = ai.Agent(tools=[delete_file, audit_log])
 
     messages = [
         ai.system_message("""

examples/temporal-direct/main.py

@@ -174,8 +174,8 @@ async def loop(self, context: ai.Context) -> AsyncGenerator[ai.events.AgentEvent
 
 def _activity_tool_call(
     tc: ai.messages.ToolCallPart,
-) -> ai.ToolCallLike:
-    """Build a ``ToolCallLike`` that runs the tool as a Temporal activity.
+) -> ai.agents.ToolCallCallable:
+    """Build a ``ToolCallCallable`` that runs the tool as a Temporal activity.
 
     ``ToolRunner.schedule`` accepts any zero-arg callable that returns
     a coroutine resolving to a ``ToolCallResult``. This lets us route

skills/ai/SKILL.md

@@ -181,17 +181,28 @@ Each forwarded event is wrapped in `ai.events.PartialToolCallResult` carrying th
 
 ## Hooks
 
-Typed suspension points for human-in-the-loop:
+Typed suspension points for human-in-the-loop.
+
+**Tool approval (built-in shortcut).** Flag a tool with `require_approval=True` and the default loop gates each call behind an `ai.tools.ToolApproval` hook (label `approve_{tool_call_id}`, payload carries `granted` + `reason`):
 
 ```python
-class Approval(pydantic.BaseModel):
-    granted: bool
-    reason: str
+@ai.tool(require_approval=True)
+async def delete_file(path: str) -> str:
+    ...
+
+# consumer-side resolve:
+ai.resolve_hook(hook.hook_id, ai.tools.ToolApproval(granted=True))
 ```
 
-Inside agent code (blocks until resolved):
+Denial returns an error `ToolCallResult` with `result=f"Rejected: {reason}"`. For custom payloads, label schemes, or per-call gating, write a custom loop using the primitives below.
+
+**Manual hooks.** Inside agent code (blocks until resolved):
 
 ```python
+class Approval(pydantic.BaseModel):
+    granted: bool
+    reason: str
+
 approval = await ai.hook(
     "approve_send_email",
     payload=Approval,

src/ai/agents/__init__.py

@@ -14,6 +14,7 @@
     SubAgentTool,
     Tool,
     ToolCall,
+    ToolCallCallable,
     ToolCallLike,
     ToolRunner,
     agent,
@@ -44,6 +45,7 @@
     "SubAgentTool",
     "Tool",
     "ToolCall",
+    "ToolCallCallable",
     "ToolCallLike",
     "ToolRunner",
     "StreamingStatusTool",