Claude Agent SDKclaude-agent-sdk 0.2.157 · Python 3.10+
Dashboard
0%
1
Curious builder0 XP earned · 300 to level 2
0 daysFinish a lesson to begin
Badge collection0 of 6 unlocked
46 small wins to finish your pathNext lesson →

Hooks: PreToolUse

A hook is a function Claude Code calls at a fixed point in the agent loop. A PreToolUse hook sees every tool call before it runs, and can let it through or stop it.

Last updated: 20 Oct, 2026 · Claude Agent SDK

An allow rule settles a call before your can_use_tool callback is asked, so a tool you listed is never checked again. Hooks sit in front of all of that: the hooks page puts the callbacks first, before matchers filter them and before any rule decides.

The files these runs use

One stand-in file carries this course. shop_desk.py holds the scripted model and the Transport that stands in for the Claude Code binary, and it arrives at this part with the tool loop and the permission steps already on it. Here it is in full, so nothing below depends on code you cannot see.

python
import asyncio
import itertools
import json
import re

from claude_agent_sdk import ClaudeAgentOptions, ProcessError, Transport

ORDER = re.compile(r"\b[A-Z]\d+\b")
SESSIONS = itertools.count(1)


class ShopModel:
    """Stands in for Claude: reads the conversation, picks the next step."""

    def __init__(self, script=None):
        self.script = list(script or [])

    def reply(self, said, results, tools):
        # a scripted reply lets a lesson force a step the stand-in would not pick
        if self.script:
            return self.script.pop(0)
        if results:
            return {"text": " ".join(results)}

        # the newest message first, then the last id named earlier in the chat
        orders = ORDER.findall(said[-1]) or ORDER.findall(" ".join(said))[-1:]
        wanted = "refund_order" if "refund" in said[-1].lower() else "lookup_order"
        tool = next((name for name in tools if name.endswith(wanted)), None)
        if orders and tool:
            return {"calls": [(tool, {"order_id": order}) for order in orders]}
        return {"text": f"I cannot help with order {orders[0]} yet." if orders
                else "Which order is this about?"}


class ShopDesk(Transport):
    """Stands in for the Claude Code binary: it runs the loop over JSON lines."""

    def __init__(self, options=None, model=None):
        self.options = options or ClaudeAgentOptions()
        self.model = model or ShopModel()

        # a counted id, not a UUID, so the printed output is the same every run
        self.session_id = f"session-{next(SESSIONS)}"
        self.inbox, self.outbox = asyncio.Queue(), asyncio.Queue()
        self.prompts, self.said = asyncio.Queue(), []
        self.ids, self.setup, self.waiting = itertools.count(1), {}, {}
        self.tools, self.failed, self.ready = {}, None, False

    async def connect(self):
        self.ready = True

    def is_ready(self):
        return self.ready

    async def write(self, data):
        # one JSON line at a time, which is what the SDK writes to a real process
        await self.inbox.put(json.loads(data))

    async def end_input(self):
        # for a real process this would close its standard input
        await self.inbox.put(None)

    async def close(self):
        self.ready = False

    async def send(self, message):
        await self.outbox.put(message)

    async def read_messages(self):
        # the listener and the worker run alongside the SDK's own read loop
        asyncio.create_task(self.listen())
        self.worker = asyncio.create_task(self.work())

        while (message := await self.outbox.get()) is not None:
            if isinstance(message, Exception):
                raise message
            yield message

    async def listen(self):
        while (message := await self.inbox.get()) is not None:
            # a control request is the SDK asking a question; initialize is first
            if message["type"] == "control_request":
                if message["request"]["subtype"] == "initialize":
                    self.setup = message["request"]
                await self.respond(message["request_id"])

            # a control response answers a question the desk itself asked
            elif message["type"] == "control_response":
                answer = message["response"]
                self.waiting.pop(answer["request_id"]).set_result(answer)
            elif message["type"] == "user":
                await self.prompts.put(message["message"]["content"])

        # the input has ended: finish the work in hand, then pass on any failure
        await self.prompts.put(None)
        await self.worker
        await self.send(self.failed)

    async def respond(self, request_id):
        # the stand-in says yes to every request the SDK makes of it
        await self.send({"type": "control_response", "response": {
            "subtype": "success", "request_id": request_id, "response": {}}})

    async def ask(self, request):
        # a future per request, so the answer can be awaited where it was asked
        request_id = f"desk_{next(self.ids)}"
        self.waiting[request_id] = asyncio.get_running_loop().create_future()
        await self.send({"type": "control_request", "request_id": request_id,
                         "request": request})
        return await self.waiting[request_id]

    async def work(self):
        while (prompt := await self.prompts.get()) is not None:
            await self.turn(prompt)

    async def start(self, prompt):
        self.said.append(prompt)
        if not self.tools:
            self.tools = await self.find_tools()
        await self.send({"type": "system", "subtype": "init", "model": "shop-desk",
                         "session_id": self.session_id, "tools": list(self.tools)})

    async def turn(self, prompt):
        await self.start(prompt)
        results, turns, limit = [], 0, self.options.max_turns
        while "calls" in (step := self.model.reply(self.said, results, self.tools)):
            turns += 1

            # max_turns is the desk's own stop, not something the model respects
            if limit and turns > limit:
                message = f"Reached maximum number of turns ({limit})"
                return await self.finish("error_max_turns", turns, message)
            results = [await self.use(name, args) for name, args in step["calls"]]

        await self.say(step["text"])
        await self.finish("success", turns + 1, step["text"])

    async def say(self, text):
        # with include_partial_messages on, the text goes out a word at a time
        if self.options.include_partial_messages:
            for word in text.split(" "):
                delta = {"type": "text_delta", "text": word + " "}
                await self.send({"type": "stream_event", "uuid": f"ev_{next(self.ids)}",
                                 "session_id": self.session_id, "event": {
                                     "type": "content_block_delta", "index": 0,
                                     "delta": delta}})

        # the whole message still follows, so code written earlier keeps working
        await self.send(self.message("assistant", [{"type": "text", "text": text}]))

    async def finish(self, subtype, turns, text):
        ok = subtype == "success"
        result = {"type": "result", "subtype": subtype, "is_error": not ok,
                  "duration_ms": 0, "duration_api_ms": 0, "num_turns": turns,
                  "session_id": self.session_id, "total_cost_usd": 0.0,
                  "usage": {"input_tokens": len(" ".join(self.said).split()),
                            "output_tokens": len(text.split())}}

        # cost is zero and usage counts words, because no model was called
        result.update({"result": text} if ok else {"errors": [text]})
        await self.send(result)
        self.failed = ProcessError("Command failed", exit_code=1) if not ok else None

    def message(self, kind, content, parent=None):
        body = {"role": kind, "content": content}
        if kind == "assistant":
            body.update(model="shop-desk", id=f"msg_{next(self.ids)}")
        return {"type": kind, "message": body, "parent_tool_use_id": parent,
                "session_id": self.session_id}

    async def mcp(self, server, method, params):
        message = {"jsonrpc": "2.0", "id": next(self.ids), "method": method,
                   "params": params}
        answer = await self.ask({"subtype": "mcp_message", "server_name": server,
                                 "message": message})
        return answer["response"]["mcp_response"]["result"]

    async def find_tools(self):
        tools = {}
        for server in self.options.mcp_servers:
            await self.mcp(server, "initialize", {"protocolVersion": "2025-06-18",
                           "capabilities": {}, "clientInfo": {"name": "desk", "version": "1"}})

            for tool in (await self.mcp(server, "tools/list", {}))["tools"]:
                name = f"mcp__{server}__{tool['name']}"

                # a tool left out here never reaches the model at all
                if name not in self.options.disallowed_tools:
                    tools[name] = (server, tool["name"])
        return tools

    async def use(self, name, args):
        call = f"toolu_{next(self.ids)}"
        block = {"type": "tool_use", "id": call, "name": name, "input": args}
        await self.send(self.message("assistant", [block]))

        allowed, args = await self.permit(name, args, call)
        text, failed = (await self.run(name, args, call)) if allowed else (args, True)

        result = {"type": "tool_result", "tool_use_id": call, "content": text,
                  "is_error": failed}
        await self.send(self.message("user", [result]))
        return text

    async def run(self, name, args, call):
        # the full name carries the server it came from, so the call goes back there
        server, tool = self.tools[name]
        result = await self.mcp(server, "tools/call", {"name": tool, "arguments": args})
        return result["content"][0]["text"], result.get("isError", False)

    async def permit(self, name, args, call):
        o = self.options
        if o.permission_mode == "bypassPermissions" or name in o.allowed_tools:
            return True, args

        # nothing approved the call and there is nobody left to ask
        if o.can_use_tool is None or o.permission_mode == "dontAsk":
            return False, f"Permission to use {name} was not granted."
        return await self.ask_permission(name, args, call)

    async def ask_permission(self, name, args, call):
        # the SDK runs your callback and writes its answer back on this channel
        decision = (await self.ask({"subtype": "can_use_tool", "tool_name": name,
                                    "input": args, "tool_use_id": call,
                                    "permission_suggestions": None,
                                    "blocked_path": None}))["response"]

        # an allow may hand back different arguments to run with
        if decision["behavior"] == "allow":
            return True, decision["updatedInput"]
        return False, decision["message"]

Two smaller files travel with it and do not change in this part: the shop's own tools, and the printer that turns a message into one readable line.

python
from claude_agent_sdk import create_sdk_mcp_server, tool

ORDERS = {"A17": "Order A17 shipped on 3 March."}


# the description is what Claude reads to decide this is the right tool
@tool("lookup_order", "Look up an order's shipping status by its id, such as A17.",
      {"order_id": str})
async def lookup_order(args):
    order = args["order_id"]

    # an unknown id is answered in words, not raised, so Claude can pass it on
    status = ORDERS.get(order, f"No order {order} was found.")
    return {"content": [{"type": "text", "text": status}]}

REFUNDED = []


@tool("refund_order", "Refund an order by its id. The money goes back to the customer.",
      {"order_id": str})
async def refund_order(args):
    REFUNDED.append(args["order_id"])
    text = f"Order {args['order_id']} has been refunded."
    return {"content": [{"type": "text", "text": text}]}

shop = create_sdk_mcp_server("shop", tools=[lookup_order, refund_order])
python
from claude_agent_sdk import (AssistantMessage, ResultMessage, SystemMessage,
                              TextBlock, ToolResultBlock, ToolUseBlock, UserMessage, query)

from shop_desk import ShopDesk


def describe(message):
    # the two messages with no content blocks are read from their own fields
    if isinstance(message, SystemMessage):
        return f"system     {message.subtype}"
    if isinstance(message, ResultMessage):
        return f"result     {message.subtype}: {message.result or message.errors}"

    # a type this course does not use prints its name rather than failing
    if not isinstance(message, (AssistantMessage, UserMessage)):
        return type(message).__name__
    return describe_block(message.content[0])


def describe_block(block):
    if isinstance(block, TextBlock):
        return f"assistant  {block.text}"
    if isinstance(block, ToolUseBlock):
        return f"assistant  calls {block.name} {block.input}"

    # a tool result sits on the user's side, and its error flag is worth seeing
    if isinstance(block, ToolResultBlock):
        flag = " (error)" if block.is_error else ""
        return f"user       result{flag}: {block.content}"
    return type(block).__name__


async def watch(prompt, options, model=None):
    async for message in query(prompt=prompt, options=options,
                               transport=ShopDesk(options, model)):
        print(describe(message))

Writing two PreToolUse hooks

A hook is an ordinary async function. The first one here only watches. The second refuses refunds and says why.

python
from claude_agent_sdk import HookMatcher


async def audit(input_data, tool_use_id, context):
    # an empty dictionary is the "no opinion" answer: the call carries on
    print("audit:", input_data["tool_name"], input_data["tool_input"])
    return {}


async def refunds_closed(input_data, tool_use_id, context):
    # the reason reaches the model, so write what a customer could be told
    return {"hookSpecificOutput": {
        "hookEventName": "PreToolUse", "permissionDecision": "deny",
        "permissionDecisionReason": "Refunds are closed until Monday."}}

Every callback takes three arguments: the event's details, the tool call's id, and a context object the Python SDK reserves for later use. It returns a dictionary. An empty one means no opinion. A permissionDecision of deny inside hookSpecificOutput blocks the call, and the hooks page lists allow, ask and defer as the other three.

Registering them with HookMatcher

hooks maps an event name to a list of matchers.

python
from claude_agent_sdk import ClaudeAgentOptions
from shop_tools import REFUNDED, shop
from watch import watch

# both tools are allowed here, so only a hook can stop the refund
options = ClaudeAgentOptions(
    mcp_servers={"shop": shop},
    allowed_tools=["mcp__shop__lookup_order", "mcp__shop__refund_order"],
    hooks={"PreToolUse": [
        HookMatcher(hooks=[audit]),
        HookMatcher(matcher="mcp__shop__refund_order", hooks=[refunds_closed]),
    ]},
)

A HookMatcher with no pattern runs its hooks for every call of that event; matcher limits them to tools whose name matches, and timeout sets how many seconds the callbacks get. Both tools are in allowed_tools here, so the rules on their own would approve a refund.

Teaching the desk to call the hooks

shop_desk.py adds three methods, hook_input, hooks and pre_hooks, and replaces permit so the hooks are asked before any rule. The file already imports re.

python
    def hook_input(self, event, data):
        # the fields a hook input carries whatever the event is, plus the
        # event's own fields on top
        return {"hook_event_name": event, "session_id": self.session_id,
                "cwd": ".", "transcript_path": "", **data}
python
    async def hooks(self, event, data):
        outputs = []
        for group in (self.setup.get("hooks") or {}).get(event, []):
            # a matcher of None means the group runs for every tool
            if group["matcher"] and not re.fullmatch(group["matcher"], data["tool_name"]):
                continue

            # the SDK sent ids, not functions, so the call goes back by id
            for callback in group["hookCallbackIds"]:
                answer = await self.ask({"subtype": "hook_callback",
                                         "callback_id": callback, "tool_use_id": data["tool_use_id"],
                                         "input": self.hook_input(event, data)})
                outputs.append(answer["response"])
        return outputs

The SDK sent your hooks in the initialize request, as callback ids grouped by event and matcher. hooks finds the groups for an event, tests the matcher, and sends a hook_callback request for each id. The SDK looks your function up by id, calls it, and sends back what it returned.

python
    async def pre_hooks(self, name, args, call):
        verdict = None
        for output in await self.hooks("PreToolUse", {"tool_name": name,
                                                      "tool_input": args, "tool_use_id": call}):
            decision = output.get("hookSpecificOutput", {})

            # one deny ends it, and the other hooks' opinions no longer matter
            if decision.get("permissionDecision") == "deny":
                return "deny", decision.get("permissionDecisionReason", "Blocked by a hook")
            verdict = decision.get("permissionDecision", verdict)
            args = decision.get("updatedInput", args)
        return verdict, args
python
    async def permit(self, name, args, call):
        # hooks are the first step, so nothing below here can skip them
        verdict, args = await self.pre_hooks(name, args, call)
        if verdict in ("allow", "deny"):
            return verdict == "allow", args

        o = self.options
        if o.permission_mode == "bypassPermissions" or name in o.allowed_tools:
            return True, args

        # nothing approved the call and there is nobody left to ask
        if o.can_use_tool is None or o.permission_mode == "dontAsk":
            return False, f"Permission to use {name} was not granted."
        return await self.ask_permission(name, args, call)

pre_hooks collects the decisions and lets any deny win, which is the order the hooks page gives: deny beats defer, defer beats ask, ask beats allow. The same page says matching hooks run in parallel and that completion order is not fixed, so each one has to stand on its own; your desk runs them one after another.

Logging every call and refusing the refund

Example
await watch("Where is A17?", options)
await watch("Please refund my order A17", options)
print("refunded:", REFUNDED)

audit saw both calls, because its matcher is empty. refunds_closed matched only the refund and denied it although the refund tool is in allowed_tools. refunded is still empty, and that list only grows inside the tool itself, so the tool never ran.

The same refund with no hook registered

Take the hook away, leave the rest alone, and whatever the next run does differently is the hook's doing.

Example
# the same two tools, the same prompt, and no hooks at all
nothing = ClaudeAgentOptions(mcp_servers={"shop": shop},
                             allowed_tools=["mcp__shop__refund_order"])

await watch("Please refund my order A17", nothing)
print("refunded:", REFUNDED)

The money moved: REFUNDED now holds A17. That is the proof the first run's empty list was worth something.

What the SDK sends over the wire

Example
from claude_agent_sdk import query
from shop_desk import ShopDesk

desk = ShopDesk(options)
async for message in query(prompt="Hello", options=options, transport=desk):
    pass
print(desk.setup["hooks"])

No functions cross the wire, only ids grouped by event and matcher. Claude Code asks for one back by id when it needs a decision, which is why a hook can live in your process and still run inside the loop.

A PreToolUse hook compared with an allow rule

allowed_toolsA PreToolUse hook
Decides bythe tool's namethe name and the arguments
Runsafter the hooksfirst, before any rule
Can rewrite the callnoyes, with updatedInput
Gives a reasonnoyes, permissionDecisionReason
Holds in bypassPermissionsnoyes, a deny still blocks

The last row is the one that catches people. The permissions page describes bypassPermissions as skipping the permission checks, and a hook deny is not one of them.

Auditing, blocking and fixing calls

  • An audit trail. One matcher-free hook that logs the tool name and arguments of every call, whatever decides on it afterwards.
  • A closed window. Refunds off outside business hours, or writes off while a release is going out, with a reason the model can repeat to the customer.
  • A limit your rules cannot express. A refund under fifty pounds allowed, anything above it denied, because the hook reads the arguments and a rule does not.
Watch out. A matcher is a pattern over the tool's name, never over its arguments. HookMatcher(matcher="/etc") matches no tool at all, so the hook silently never fires and you believe a path is guarded when it is not. Match on the tool, then read tool_input inside the callback.
Try it yourself
  • Change the second matcher to "mcp__shop__.*" and ask about A17. Which call is denied now?
  • Add permission_mode="bypassPermissions" to the options and run the refund again. Read REFUNDED.
  • Return {"systemMessage": "checked"} from audit instead of {} and see whether the call still runs.
PreviousTurn and budget limits

You understood something today that you didn't yesterday.