Task 5 · 8 tasks

Hooks and interrupts

Pause the agent for human approval before any destructive SQL reaches the database.

25 minMedium

Alice’s ask

The “Rogue Agent” problem

During testing the agent cheerfully ran a DELETE. Alice wants a human in the loop: reads are fine, anything that changes data needs explicit approval.

You

Build it

Write an approval hook

Challenge

Finish ApprovalHook: before any query_db call whose SQL contains DELETE, UPDATE, INSERT, DROP, ALTER or TRUNCATE, pause and ask a human. Anything but y cancels the call.

Work in phase1/starter/t5_hooks_interrupts.py (look for TODO; an unfinished one prints [starter] TODO …) and run it with uv run bootcamp.py phase1 t5 --starter. The reference solution is phase1/hooks.py; uv run bootcamp.py phase1 t5 runs it.

Hint 1

Hooks are typed lifecycle callbacks. You want BeforeToolCallEvent, registered from a HookProvider. The event can interrupt() the agent and can cancel the tool. See Strands: hooks and interrupts.

Hint 2

Read the tool name from event.tool_use["name"] and the SQL from event.tool_use["input"]["query"]. The interrupt's return value is the human's answer.

pseudocoderead only

def approve(self, event):
    if <not query_db>: return
    if <destructive>:
        answer = event.interrupt("approval-required", reason={"query": query})
        if answer != "y": event.cancel_tool = "..."

Solution

phase1/starter/t5_hooks_interrupts.py (keyword version)

class ApprovalHook(HookProvider):
    def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
        registry.add_callback(BeforeToolCallEvent, self.approve)

    def approve(self, event: BeforeToolCallEvent) -> None:
        if event.tool_use.get("name") != "query_db":
            return
        query = event.tool_use["input"].get("query", "")
        if any(word in query.upper() for word in DESTRUCTIVE_KEYWORDS):
            approval = event.interrupt("approval-required", reason={"query": query})
            if str(approval).strip().lower() != "y":
                event.cancel_tool = "User denied permission to run this query"

The reference goes further: instead of searching for keywords it asks shared/sql_guard.py, an allowlist that only treats SELECT, EXPLAIN, read-only WITH queries and schema PRAGMAs such as PRAGMA table_info(employees) as reads, ignoring words inside string literals and comments:

phase1/hooks.py

class ApprovalHook(HookProvider):
    """Interrupts any destructive `query_db` call until a human answers 'y'."""

    def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
        """Run `approve` before every tool call."""
        registry.add_callback(BeforeToolCallEvent, self.approve)

    def approve(self, event: BeforeToolCallEvent) -> None:
        """Ask for approval when a `query_db` call may write; cancel the call unless the answer is 'y'."""
        if event.tool_use.get("name") != "query_db":
            return
        query = event.tool_use["input"].get("query", "")
        if sql_guard.is_write(query):
            approval = event.interrupt("approval-required", reason={"query": query})
            if str(approval).strip().lower() != "y":
                event.cancel_tool = "User denied permission to run this query"

Answer the interrupt
Challenge
In the same starter file, when the agent stops for an interrupt, ask the human and resume the agent with the answers (make_responder).
Hint 1
An interrupted run returns with result.stop_reason == "interrupt" and a list of result.interrupts, each with an id, name and reason.
Hint 2
Resume by calling the agent again with a list of {"interruptResponse": {"interruptId": ..., "response": ...}}. Loop: one answer may trigger another interrupt.
Solution
phase1/t5_hooks_interrupts.py
def make_responder(agent: Agent): def respond(prompt: str): result = agent(prompt) while result.stop_reason == "interrupt": responses = [ {"interruptResponse": {"interruptId": i.id, "response": ask_approval(i.reason["query"])}} for i in result.interrupts if i.name == "approval-required" ] result = agent(responses) return result return respond
Try it, safely
terminal · your starter file
```
uv run bootcamp.py phase1 t5 --starter "Delete employee with ID 5"
```
terminal · reference solution
```
uv run bootcamp.py phase1 t5 "Delete employee with ID 5"
```
One-shot runs auto-deny destructive queries; leave out the prompt to be asked y/N.
The database is your own copy, so approving a delete only affects you. Delete phase1/datastream_corp.db and re-run uv run bootcamp.py phase1 t0 for a fresh one.
Experiments
- Keyword matching is crude. Ask “Show the updated_at column of employee 5” with the keyword version: updated_at contains UPDATE, so a harmless read asks for approval. Can you phrase a request that changes data but slips past the list? Compare with the reference (sql_guard).
- Why does the deployed agent in Phase 2 block instead of asking? Who would answer?

Check your work

Phase 1 has no automated test: you check it by running the task and looking for the result below.

“Delete employee with ID 5” pauses and asks for approval. Answering y runs it; anything else cancels it and the agent reports that permission was denied. With the reference guard, reads (SELECT, EXPLAIN, read-only WITH, and schema PRAGMAs such as PRAGMA table_info(employees)) run without a prompt; the keyword version may also stop harmless reads that mention a keyword, such as updated_at.

Under the hood

Hooks are typed lifecycle callbacks (before/after model call, before/after tool call, and more). event.interrupt() suspends the agent loop and surfaces a named interrupt to the caller; the response you send back becomes the return value inside the hook. Setting event.cancel_tool skips the tool and returns your message to the model as the tool result.