---
title: Human in the Loop
description: Add approvals and external decisions to agent runs.
type: guide
summary: Use tool approvals, manual hooks, cancellation, denial, and serverless resume flows.
---

# Human in the Loop



Tool approval in the SDK is a high-level API built on top of hooks. Hooks
suspend the loop until your application resolves them externally with some
payload.

## Require tool approval

Pass `require_approval=True` to `@ai.tool` when a tool needs approval:

```python
@ai.tool(require_approval=True)
async def notify_mothership(message: str) -> str:
    """Notify the mothership."""
    return f"Sent: {message}"
```

When the model calls the tool, the agent emits a hook event. Resolve it with
`ai.resolve_hook`:

```python
async with agent.run(model, messages) as stream:
    async for event in stream:
        if (
            isinstance(event, ai.events.HookEvent)
            and event.hook.status == "pending"
        ):
            print(event.hook.hook_id, event.hook.metadata)
            ai.resolve_hook(
                event.hook,
                ai.tools.ToolApproval(granted=True, reason="approved"),
            )
```

Return a denial by resolving the hook with `granted=False`:

```python
ai.resolve_hook(
    "approve_tool_call_id_here",
    ai.tools.ToolApproval(granted=False, reason="not allowed"),
)
```

Cancel a live hook when the waiting workflow should stop:

```python
await ai.cancel_hook("approve_tool_call_id_here", reason="client disconnected")
```

Hook operations target the current `HookRegistry`, which is set inside the
`async with agent.run(...)` block. To resolve from a different task — a UI
callback, another endpoint — pass a registry explicitly. Create one up front
and hand it to the run (or use the run's own via `stream.hook_registry`):

```python
registry = ai.HookRegistry()

async with agent.run(model, messages, hook_registry=registry) as stream:
    ...

# elsewhere, e.g. in a UI callback:
ai.resolve_hook(hook_id, approval, registry=registry)
```

`ai.get_hook_registry()` returns the current registry (raising `LookupError`
when there is none), which is handy to capture the run's registry without
threading `stream` around:

```python
async with agent.run(model, messages) as stream:
    on_approval_needed(registry=ai.get_hook_registry())
    ...
```

To make a registry current for a block of code outside a run — so plain
`resolve_hook(...)` calls inside it need no `registry=` argument — use
`ai.agents.hooks.use_hook_registry`. It is an async context manager and
also works as a decorator on async functions:

```python
from ai.agents import hooks

async with hooks.use_hook_registry(registry):
    ai.resolve_hook(hook_id, approval)

@hooks.use_hook_registry(registry)
async def on_decision(hook_id: str, approval: ai.tools.ToolApproval) -> None:
    ai.resolve_hook(hook_id, approval)
```

## Build custom hooks

Use `ai.hook` to define your own custom hooks to suspend execution
and deliver external information into the loop:

```python
class CustomPayload(pydantic.BaseModel):
    foo: int

approval = await ai.hook(
    "some_hook",
    payload=CustomPayload,
    metadata={"kek": "pek"},
)
```

Resolve the hook from another part of your application:

```python
ai.resolve_hook(
    "some_hook",
    {"foo": 999},
)
```

## Resume in serverless flows

Serverless setups can't suspend on `await` in the same way as long-running
servers. For that reason, the SDK provides an alternative way for handling
hook resolutions.

Using the serverless flow only requires you to update the `agent.run` callsite
(e.g. your serverless endpoint code).

```python
# start (or replay pre-hook part of) a run
async with agent.run(model, messages) as stream:
    # pre-register tool approvals before iterating the stream
    for approval in approvals:
        ai.resolve_hook(
            approval.hook_id,
            ai.tools.ToolApproval.model_validate(approval.data)
        )

    async for event in stream:
        if (
            isinstance(event, ai.events.HookEvent)
            and event.hook.status == "pending"
        ):
            # interrupt the loop
            ai.defer_hook(event.hook)

```

When handling the hook event, call `defer_hook` to interrupt the loop
and surface your tool approvals (or custom hook-related work) to the client.

The `ToolRunner` will handle multiple concurrent aborts by waiting for all
scheduled tasks to complete or get aborted. That way you can run approval-gated
tools on serverless concurrently.

Once the client has gathered payloads for all hooks, it will re-enter via the
same endpoint, which will then pre-register hook resolutions inside the
`agent.run` block, before iterating the stream. The loop will replay results
of LLM calls and tool results
from the first run without redoing non-deterministic and duplicating side-effects.
When it hits the hook for which it has a pre-registered resolution, it will
continue through without suspending.


---

For a semantic overview of all documentation, see [/sitemap.md](/sitemap.md)

For an index of all available documentation, see [/llms.txt](/llms.txt)

For agent-facing discovery, including API and MCP surfaces, see [/agents.md](/agents.md)