Hooks that change a call

A hook output is the dictionary a hook returns to change what happens next. A PreToolUse hook can rewrite a call's arguments and approve it, and a PostToolUse hook can rewrite the answer before the model reads it.

Last updated: 20 Oct, 2026 · Claude Agent SDK

The PreToolUse hooks so far either let a call through or stopped it. The same return value can repair a call on its way in, and a second event can edit the result on its way out.

Fixing the input and rewriting the output

python

async def tidy_order_id(input_data, tool_use_id, context):
    # the model sent a messy id: fix it here rather than in every tool
    order = input_data["tool_input"]["order_id"].strip().upper()
    return {"hookSpecificOutput": {
        "hookEventName": "PreToolUse", "permissionDecision": "allow",
        "updatedInput": {**input_data["tool_input"], "order_id": order}}}


async def after(input_data, tool_use_id, context):
    # PostToolUse runs once the tool has answered, so it sees tool_response
    print("after:", input_data["tool_input"], "->", input_data["tool_response"])
    return {"hookSpecificOutput": {"hookEventName": "PostToolUse",
                                   "updatedToolOutput": "Shipped on 3 March."}}

tidy_order_id returns updatedInput, the arguments to run with instead. The hooks page asks for a new dictionary rather than a changed one, and says that pairing it with allow approves the rewritten call, while leaving permissionDecision out lets the new input go through the normal permission check. after is a PostToolUse hook: its input has tool_response, and updatedToolOutput replaces what the model is about to read.

Calling PostToolUse from the desk

run in shop_desk.py is replaced with a version that calls the PostToolUse hooks once the tool has answered, and takes their replacement text.

python

    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})

        # the hooks see the answer before the model does, with the same call id
        text = result["content"][0]["text"]
        for output in await self.hooks("PostToolUse", {"tool_name": name, "tool_input": args,
                                                       "tool_response": text, "tool_use_id": call}):
            # updatedToolOutput replaces what the model is about to read
            decision = output.get("hookSpecificOutput", {})
            text = decision.get("updatedToolOutput", text)
        return text, result.get("isError", False)

The call id is the same one the PreToolUse hooks saw, which is how a pair of hooks can match a request to its answer.

Project files used on this pageThis lesson builds on a project from earlier lessons. The code below imports these files. Click a file to see its code, or follow the link to the lesson that wrote it. To run the code yourself, keep them in the same folder.

View the code here

watch.py

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))

shop_desk.py

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})

        # the hooks see the answer before the model does, with the same call id
        text = result["content"][0]["text"]
        for output in await self.hooks("PostToolUse", {"tool_name": name, "tool_input": args,
                                                       "tool_response": text, "tool_use_id": call}):
            # updatedToolOutput replaces what the model is about to read
            decision = output.get("hookSpecificOutput", {})
            text = decision.get("updatedToolOutput", text)
        return text, result.get("isError", False)

    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}

    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

    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

    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)

    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"]

shop_tools.py

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])

A model that sends a messy id

The permission callback prints when it is asked, so the run shows whether it was reached.

python

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher, PermissionResultAllow
from shop_tools import shop
from watch import watch


async def check(tool_name, input_data, context):
    # printed so the run shows whether the callback was reached at all
    print("check was asked")
    return PermissionResultAllow()


options = ClaudeAgentOptions(mcp_servers={"shop": shop}, can_use_tool=check, hooks={
    "PreToolUse": [HookMatcher(matcher="mcp__shop__lookup_order", hooks=[tidy_order_id])],
    "PostToolUse": [HookMatcher(hooks=[after])]})

The model is scripted to ask for " a17 ", with spaces and a small letter, which is the kind of argument a real model produces when an order id was typed into a chat box.

Example

from shop_desk import ShopModel

messy = {"calls": [("mcp__shop__lookup_order", {"order_id": " a17 "})]}
sloppy = ShopModel(script=[messy])
await watch("Where is my order?", options, model=sloppy)

Output

system     init
assistant  calls mcp__shop__lookup_order {'order_id': ' a17 '}
after: {'order_id': 'A17'} -> Order A17 shipped on 3 March.
user       result: Shipped on 3 March.
assistant  Shipped on 3 March.
result     success: Shipped on 3 March.

The call went out as ' a17 ' and the tool still found the order, because it ran with the hook's arguments, not the model's.
after printed {'order_id': 'A17'}, the tidied input, beside the tool's real answer.
The model read Shipped on 3 March., which no tool ever returned. That is updatedToolOutput standing in front of the result.
check was never asked. A PreToolUse allow settles the call, so the permission callback is not reached.

Which field changes what

Field	Event	What it changes
updatedInput	PreToolUse	the arguments the tool runs with
permissionDecision	PreToolUse	allow, deny, ask or defer
permissionDecisionReason	PreToolUse	the text the model is given
updatedToolOutput	PostToolUse	the result, for any tool
additionalContext	PostToolUse	text appended to the result
systemMessage	any event	a line for the person, not the model

updatedMCPToolOutput is an older field that only covers MCP tools; the hooks page says to prefer updatedToolOutput, which works for all of them.

Repairing, redacting and recording

Normalising arguments. Trim and upper-case an id, expand a relative path, add the account the customer is signed in under.
Redacting an answer. A PostToolUse hook that strips a card number out of a tool's result so it never reaches the model's context.
Writing an audit row. The PreToolUse and PostToolUse inputs share a tool_use_id, so one row can hold the request and its answer.

Watch out. Returning allow is a decision, not a hint: it skips your can_use_tool callback. The permissions page adds that deny rules and ask rules still apply afterwards. Return an allow only for a call you have checked yourself.

Previous: Hooks: PreToolUse
Next: The hook events
Reference: Intercept and control agent behavior with hooks

Try it yourself

Delete "permissionDecision": "allow" from tidy_order_id and run it again. Who is asked now, and does the tidied id still reach the tool?
Return additionalContext instead of updatedToolOutput from after and read what the model answers.
Give after the matcher mcp__shop__refund_order and ask about A17 again.

PreviousHooks: PreToolUse

Slow is fine. Stopping is the only problem.

Hooks that change a call

Fixing the input and rewriting the output

Calling PostToolUse from the desk

A model that sends a messy id

Which field changes what

Repairing, redacting and recording

Related