Hooks and Real-Time Steering¶

Hooks let a host application observe and steer an agent run while it is happening — veto a tool call before it executes, inject a message mid-run, pause for a human confirmation, or abort cleanly.

A hook is an async (or plain sync) callable handed to Agent(hooks=...). The toolkit awaits it at defined points in the tool-calling loop, hands it an AgentStreamEvent, and acts on the HookDirective it returns.

from doc_processing_toolkit.agent import Agent, HookDirective

async def guard(event):
    if event.event_type == "before_tool_call" and event.metadata["tool_name"] == "delete_file":
        return HookDirective(action="skip_tool", message="File deletion is disabled.")
    return None  # everything else proceeds untouched

agent = Agent(model="claude-sonnet-5-5", tools=[...], hooks=guard)
result = await agent.run("Clean up the temp directory")

Import path

The hook types are exported from the agent subpackage, not the top-level package: from doc_processing_toolkit.agent import HookDirective, AgentAborted, AgentHooks, RunHandle, ConfirmAfterNToolCalls. Agent itself is available from either.

Hooks vs. stream_callback¶

The toolkit has two channels into a running agent. They share emission points internally but have deliberately different contracts, and nothing is deprecated — use either or both.

	`stream_callback` (via `Agent.stream()`)	`hooks`
Contract	Synchronous, fire-and-forget observation tap	Awaited control channel
Blocking	Never blocks the loop	The loop genuinely stops until your coroutine returns
Return value	Ignored	A `HookDirective` that changes what happens next
Text/thinking deltas	Yes	No — see below
Can veto a tool call	No	Yes (`skip_tool`)

Why deltas stay on the callback

text_delta and thinking_delta are not routed to hooks. Awaiting a hook once per token would stall streaming. If you need token-level output, use Agent.stream() — a run can have a streaming consumer and a hook at the same time.

Event vocabulary¶

These are the events delivered to the hooks channel, in the order they fire:

Event	Fires	`metadata`
`run_start`	Once, before the first API call	`{"model": str, "provider": str}`
`iteration_start`	Before each request of the tool-calling loop (a roundtrip announces its number twice)	`{"iteration": int, "loop_counter": int}`
`tool_call_start`	A tool call has been requested by the model	`{"tool_name": str, "tool_call_id": str, "arguments": dict, "iteration": int}`
`before_tool_call`	Veto point — immediately before dispatch	Same as `tool_call_start`
`tool_call_result`	A tool call finished (or was declined)	`{"tool_name": str, "tool_call_id": str, "result": str, "result_truncated": bool, "success": bool, "duration_ms": int, "iteration": int}`; `result_data` too when `tool_result_preview_chars=0`, and `tool_reported_cost` / `tool_reported_cost_estimated` when the tool reported spend. `result` is the first 500 characters by default; see Tool results in events
`iteration_complete`	A tool roundtrip finished (its follow-up response arrived)	`{"iteration": int, "input_tokens": int, "output_tokens": int, "cost": float, "tools_called": list, "stop_reason": str}`
`run_end`	Once, after the result is assembled	`{"iterations": int, "input_tokens": int, "output_tokens": int, "cost": float, "error": str\\|None}`
`request_start`	Opt-in — immediately before each provider request	`{"request_index": int, "iteration": int, "attempt": int, "purpose": str, "model": str, "provider": str}`

How iterations are numbered

An iteration is one tool roundtrip: the request whose response asks for tools, plus the follow-up request that answers them. iteration_start fires once per request, so a roundtrip announces its number twice — before its request and again before its follow-up — and request_start carries the same number. Every event of a roundtrip carries that one number: its tool_call_start, before_tool_call and tool_call_result events, and the roundtrip's iteration_complete, which fires once the follow-up's response has arrived and lists the tools in tools_called. A follow-up response that asks for more tools opens the next iteration, so those tool events carry the next number.

One tool round then a final answer, on all four providers:

iteration_start      iteration=1
tool_call_start      iteration=1
before_tool_call     iteration=1
tool_call_result     iteration=1
iteration_start      iteration=1   # the follow-up request
iteration_complete   iteration=1   tools_called=["lookup_invoice"]

The final answer does not emit an iteration_complete of its own after a tool round; a run that calls no tools emits exactly one, with tools_called=[].

An iteration_complete event's input_tokens, output_tokens and cost cover every model request of the tool loop billed since the previous event (0.5.1; each loop adds every request it prices to a utils.IterationEventUsage, and each event takes what accumulated). The first event of a tool run carries the request that asked for the tools as well as its follow-up, and a pause_turn continuation or a max_tokens recovery request is in the next event. So a run's events sum to its per-request records up to its last event. The final result is the only report of:

a request after the last event, such as Gemini's empty-turn re-request after the final round;
a run that ends without an event, such as a forced tool_choice exit, a Path B structured answer on the first request, or an error on the first request.

input_tokens is therefore billed input summed across those requests, not the size of the latest prompt: the first event of a tool run counts two prompts. Read iteration_metrics for one request's prompt size. Compaction summarisation calls are not requests of the loop; they are on compaction_cost. Until 0.5.1 an event carried the closing request alone, so the request that asked for the tools was in no event; there is no switch back.

Before 0.4.0 the three tool events reported the number of roundtrips already closed — 0 in the first round — one lower than every other event of their roundtrip.

request_start is the one observation-only hook point

It fires only under Agent(emit_request_start=True), and only on the tool-calling path — the structured-output and plain-text path takes no hooks. A directive other than continue is ignored with a warning there, because the loop is between a completed safe point and a send: there is no pending tool call to veto and no legal place to insert a message. Steer from before_tool_call, tool_call_result or iteration_complete instead.

The RunHandle pause/abort channel still applies at that point — that is the host speaking, not the hook's return value, and request_start is not allowed to be the one place an aborting host is ignored. The set is agent.hooks.OBSERVATION_ONLY_EVENTS; the full payload and the purpose vocabulary are documented in Streaming.

tool_call_start fires before before_tool_call

Both carry identical metadata. tool_call_start is the announcement (mirroring the streaming event of the same name); before_tool_call is the decision point whose return value is honored. Put veto logic in before_tool_call — a skip_tool returned from tool_call_start is ignored with a warning.

The veto covers toolkit-dispatched tools only

before_tool_call fires for tools the toolkit dispatches — decorated @tool functions and local/remote MCP tools. Provider-native built-in tools (web search, code execution) run server-side inside the model's own turn and never surface as a vetoable dispatch. See Built-in Tools.

HookDirective¶

@dataclass
class HookDirective:
    action: Literal["continue", "skip_tool", "inject_message", "pause", "abort"] = "continue"
    message: Optional[str] = None
    reason: Optional[str] = None

Action	Valid at	Effect
`continue`	Anywhere	The loop proceeds exactly as if no hook existed. Returning `None` is equivalent.
`skip_tool`	`before_tool_call` only	The tool is not executed; its slot is answered with a synthetic declined result.
`inject_message`	Anywhere	Queues `message` as a user-role message, applied at the next turn boundary.
`pause`	Anywhere	Blocks the run at the next safe point until `run_handle.resume()`.
`abort`	Anywhere	Raises `AgentAborted`.

A hook steers the loop; it does not interrupt it¶

Every hook point sits at a boundary the loop was going to reach anyway, and a directive is honoured when the loop next arrives at one. The practical consequence is worth stating plainly: nothing in this API pre-empts a tool that is already executing.

skip_tool is the veto, and it is offered at before_tool_call — before the tool starts — which is the only moment a tool call can still be stopped for free. pause and abort are requests: the in-flight tool runs to completion, its result is appended (every provider requires a result for every tool call it emitted), and only then does the run stop. RunHandle.inject behaves the same way, which is why its drain latency is bounded by your longest-running tool rather than by the loop.

So a host that needs a hard stop in the middle of a long tool has to build it into the tool itself — a timeout, a cancellation token, a bounded subprocess. That is deliberate rather than an omission: cancelling a tool mid-flight would leave the model with an unanswered tool slot, which Anthropic rejects outright with a 400 and the other providers handle inconsistently. "Clean abort" means exactly that no provider call is cut off mid-request and no tool call is left without its result.

continue is the normal case¶

Most hooks are pure side effects — logging, metrics, progress UI, database writes. Return None (or HookDirective()) and nothing about the run changes. No synthetic content is created.

skip_tool answers the slot it declines¶

Every provider requires each tool call the model emitted to be answered by a matching tool result. Simply not executing the tool would leave an orphaned call, which the APIs reject outright (Anthropic returns a 400). So a veto automatically supplies a synthetic result in that slot:

return HookDirective(action="skip_tool", message="Deletion requires an admin approval ticket.")

The model reads that text and adapts — it picks another approach, asks the user, or finishes without the tool. Without message, the text defaults to "Tool call declined by the host application." You never construct this result yourself; message only customizes its wording.

inject_message: the three-state contract¶

inject_message queues a user-role message. It is never applied where the directive was returned — a message cannot legally sit between a tool call and its result — so what happens next depends on whether the run reaches another turn boundary. There are exactly three outcomes, and every queued message ends in one of them:

Outcome	When	Where you find it
Delivered	Another turn boundary follows: the loop appends it after the tool results and before the next API call, the same seam compaction uses.	Inline in `result.messages`, and listed in order in `result.injected_messages`.
Surfaced as undelivered	The run ends first — the model was composing its final answer, or it had already called the final-output tool.	`result.undelivered_injections`, and `complete.metadata["undelivered_injections"]` on a stream.
Attached to the exception	The run raised (an abort, or a provider error that propagated).	`exc.undelivered_injections` on the raised exception, and on the stream's terminal `complete` event.

The message is built in provider-native shape (build_injected_user_message), so the same hook code works on all four providers.

result = await agent.run("Summarise the contract")

result.injected_messages        # ["Focus on the indemnity clauses"] — the model saw this
result.undelivered_injections   # ["Actually, only section 7"]      — it did not

Both lists are flat, ordered and disjoint: a text appears in exactly one of them. That matters, because the natural thing to do with an undelivered message is to send it as the next turn — and if the toolkit had also recorded it as delivered, it would be counted twice in the conversation.

A late message becomes the next turn¶

The window for the third case is not milliseconds. It is the whole final generation — often 20–60 seconds on a long summary. A user who types during it has their message accepted by your UI and then, before this release, silently destroyed: there was no field, no event and no warning.

The toolkit deliberately does not extend the run to deliver it. Appending a user message after the model's final answer is structurally a new turn, not a continuation of the old one, so the toolkit hands the text back and lets you send it:

result = await session.run("Summarise the contract")

if result.undelivered_injections:
    # An ordinary next turn — nothing special about it.
    result = await session.run("\n\n".join(result.undelivered_injections))

A clean run with undelivered text logs one INFO line. A run that raised with text still queued logs a WARNING, because that is the one case where a caller who never inspects the exception can still lose it.

A session can take that decision for you. agent.session(late_injection_policy="new_turn") runs the extra turn itself, bounded by max_follow_up_turns and refused after an error, an abort or a repeated text. Each turn stays an ordinary turn with its own request, result and cost records, so nothing about the billing becomes approximate — see Sessions. The default, "surface", is the behaviour above.

Do not drain your inbox at tool_call_result and assume a drain follows

It is tempting to assume that queueing at tool_call_result guarantees delivery, because a drain normally comes next. It does not. Under a forced tool_choice — which is how every schema-driven agent ends its runs — all four provider loops exit between appending the tool result and reaching the drain. The message is queued and the run returns. This is exactly why the toolkit surfaces undelivered_injections: a client cannot work around a loss it is never told about.

Controlling a live run with RunHandle¶

For the duration of a hooked run, agent.run_handle exposes a RunHandle:

agent.run_handle.paused      # bool
agent.run_handle.pause()     # pause at the next safe point
agent.run_handle.resume()    # release a paused run
agent.run_handle.abort("user cancelled")   # clean abort (also releases a pause)

run_handle is None outside a hooked run — it is set at run start and cleared at run end, including when the run ends in an abort or a provider error, so a failed run never leaves a stale handle exposed. A host that needs to control a run should capture the handle from the task running the agent.

resume() and abort() are safe to call from any thread. A run started with run_sync blocks its caller inside asyncio.run, so the host necessarily releases it from another thread (a request handler, a queue consumer); the wake-up is marshalled onto the run's own event loop, so the paused run actually continues rather than waiting for unrelated I/O to wake it.

`RunHandle.inject` steers a run from outside a hook¶

agent.run_handle.inject("Actually, only the indemnity clauses")

inject(text) appends to the same queue an inject_message directive appends to, from any thread, so a host no longer needs a hook that polls its own inbox on every event. The delivery rule is identical either way: the text is delivered at the run's next turn boundary if one comes, and returned on AgentResult.undelivered_injections if the run ends first.

It does not pre-empt a tool that is already running — drain latency is bounded by your longest-running tool — and calling it after the run has finished raises RuntimeError rather than accepting a message nobody will read. Empty or non-string text raises ValueError. The queue is lock-guarded and closes when the end-of-run capture takes it, so neither channel can race a message into a run that has already ended.

Pause is in-process only (v1)

pause is an asyncio.Event in the running process's memory. That means:

A paused run does not survive a process restart. Conversation history persists only up to the last completed turn that session persistence saved; mid-turn progress is lost.
A paused run holds its resources — the coroutine, provider client state, and any MCP subprocess sessions stay alive. Pausing hundreds of runs holds hundreds of contexts.
Pause does not survive serialization; you cannot pause on one worker and resume on another.

Run agents inside a long-lived worker or background task (not a request handler that times out), keep the handle in that process, and apply a timeout policy — auto-abort or auto-resume after N minutes — so an unanswered confirmation cannot hold resources forever.

AgentAborted¶

abort raises AgentAborted out of agent.run(). It propagates cleanly through every provider loop's exception handling rather than degrading into a generic failed-run result.

from doc_processing_toolkit.agent import AgentAborted

try:
    result = await agent.run("Process the backlog")
except AgentAborted as exc:
    print(exc.reason)    # the hook-supplied rationale, or None
    print(exc.partial)   # {"input_tokens", "output_tokens", "cost", "iterations"}
    print(exc.undelivered_injections)   # queued messages the model never saw

undelivered_injections is the third state of the inject_message contract. run_end does not fire when an exception propagates, so without this attribute a message queued at the final boundary of a run that then aborted would be destroyed along with the hook runner. It is an empty list unless something really was queued — and any exception raised out of a hooked run carries it, not just AgentAborted.

partial is best-effort

The loops feed partial from each completed API request (tokens and token cost) and each completed iteration, so an abort mid-run reports what was actually consumed up to the last completed request, plus the spend of tools that had run (tracked before their tool_call_result event, so an abort there includes it). Built-in tool costs and the request in flight at the abort are not included — for exact figures, read AgentResult from a non-aborted run.

Under stream() an abort is an event, not an exception

Agent.stream() and AgentSession.stream() never raise. An abort surfaces as an error event whose metadata carries aborted: True, the reason, and the same partial figures, followed by the terminal complete event every stream ends with — so a session banks the aborted turn's spend. See Streaming. Before 0.3.0 the error event zeroed an abort's usage and no complete event followed.

Attaching hooks¶

Three forms are accepted, all normalized internally:

Async functionPlain sync functionAgentHooks object

async def my_hook(event):
    ...
    return None

agent = Agent(model="gpt-5.6-sol", hooks=my_hook)

def my_hook(event):
    log.info(event.event_type)
    return None

agent = Agent(model="gpt-5.6-sol", hooks=my_hook)

Sync callables run via asyncio.to_thread, so a slow hook cannot block the event loop.

class MyHooks:
    def __init__(self):
        self.tool_count = 0

    async def on_event(self, event):
        if event.event_type == "tool_call_result":
            self.tool_count += 1
        return None

agent = Agent(model="gpt-5.6-sol", hooks=MyHooks())

An object is the right form when the hook needs state — counters, inboxes, budget trackers. AgentHooks is a runtime_checkable Protocol: any object with an async on_event satisfies it.

A crashing hook never kills the run

If your hook raises, the exception is logged with a full traceback and the run continues as if the hook had returned continue. The one exception is AgentAborted, which is re-raised — so raising it from inside a hook is a valid way to abort.

Patterns¶

Pure side-effect hook¶

The most common hook: observe, record, change nothing.

class ProgressReporter:
    """Push tool activity to the UI without altering the run."""

    def __init__(self, publish):
        self.publish = publish

    async def on_event(self, event):
        if event.event_type == "tool_call_start":
            await self.publish(f"Running {event.metadata['tool_name']}…")
        elif event.event_type == "tool_call_result":
            ok = event.metadata["success"]
            await self.publish(f"{event.metadata['tool_name']} {'ok' if ok else 'failed'}")
        return None

agent = Agent(model="claude-sonnet-5-5", tools=[...], hooks=ProgressReporter(publish))

Confirm after N tool calls¶

The "this agent has been working for a while — continue?" pattern ships as a built-in. ConfirmAfterNToolCalls counts tool_call_result events (not iterations — one iteration can execute many tools) and pauses at the following safe point. The count restarts at every run_start, so one instance can be shared across an agent's runs without carrying calls over from the previous run.

from doc_processing_toolkit.agent import ConfirmAfterNToolCalls

confirm = ConfirmAfterNToolCalls(5)
agent = Agent(model="claude-haiku-4-5", tools=[...], hooks=confirm)

task = asyncio.create_task(agent.run("Audit every invoice in the queue"))

# Elsewhere in the same process, once the user answers:
if agent.run_handle and agent.run_handle.paused:
    agent.run_handle.resume()          # or .abort("user declined")

result = await task

Resuming is left to the host: deliver the user's answer back to the process holding the handle (websocket, queue, DB poll — your choice) and apply a timeout policy so an unanswered confirmation does not hold resources forever.

Inbox injection¶

Give a running agent new information without restarting the run. The host writes into an inbox; the agent's hook drains it and returns inject_message.

class Inbox:
    """Deliver host-supplied messages into a live run."""

    def __init__(self):
        self._pending: list[str] = []

    def post(self, text: str) -> None:
        self._pending.append(text)

    def drain(self) -> list[str]:
        pending, self._pending = self._pending, []
        return pending

    async def on_event(self, event):
        if event.event_type != "tool_call_result":
            return None
        pending = self.drain()
        if not pending:
            return None
        return HookDirective(action="inject_message", message="\n".join(pending))

inbox = Inbox()
session = Agent(model="claude-sonnet-5-5", tools=[...], hooks=inbox).session()

task = asyncio.create_task(session.run("Draft the quarterly summary"))
inbox.post("Correction: Q3 revenue was 4.2M, not 3.8M.")
result = await task

# The run may have ended before a boundary came. Deliver the remainder as the
# next turn — the toolkit will not do it behind your back.
if result.undelivered_injections:
    result = await session.run("\n\n".join(result.undelivered_injections))

A delivered message arrives as a user-role message at the next turn boundary, so the model treats it exactly as if the user had typed it.

The follow-up is not optional

Draining at iteration_complete or tool_call_result does not guarantee delivery. The last iteration_complete of a run is followed by the loop's exit, not by a drain; and under a forced tool_choice the exit sits between the tool-result append and the drain in all four provider loops. Without the undelivered_injections follow-up above, an inbox message posted during the final answer is accepted by your UI and then dropped. The toolkit tells you when this happened (INFO on a clean run, WARNING when the run raised) — acting on it is yours.

Drain on a boundary event

Hook into tool_call_result or iteration_start rather than before_tool_call. A directive returned from before_tool_call is queued anyway, so draining on a boundary event keeps the timing obvious.

Sub-agent routing — the host is the message bus¶

Hooks plus A2A sessions compose into multi-agent messaging. The toolkit deliberately does not broker messages between sessions it does not own — routing and trust policy stay with the host, which decides who may talk to whom.

class Router:
    """Route one session's output into another session's inbox."""

    def __init__(self, name, bus):
        self.name = name
        self.bus = bus            # host-owned {session_name: Inbox}

    async def on_event(self, event):
        # Outbound: report finished work to the main agent.
        if event.event_type == "run_end" and self.name != "main":
            self.bus["main"].post(f"Sub-agent {self.name} reports: {event.metadata}")

        # Inbound: drain anything the host routed to me.
        pending = self.bus[self.name].drain()
        if pending:
            return HookDirective(action="inject_message", message="\n".join(pending))
        return None

The same shape covers all three directions:

Main → sub — between turns, call sub_session.run(message) directly; mid-turn, write into the sub-agent's inbox and let its hook inject.
Sub → main — the sub-agent's hook (or its final result) posts to a queue the host routes into the main agent's inbox.
Sub ↔ sub — identical; any session's outbound event can be routed into any other session's inbox.

This works the same whether the sessions live in the toolkit's A2A registry or in the host's own process, which is what makes a host-side sub-agent implementation and the built-in create_agent_session tools interchangeable.

Approving browser and desktop actions — before_tool_call hooks that pause, refuse and audit the std-tools browser and computer tools
Streaming — the passive observation channel and Agent.stream()
MCP — A2A session tools for spawning persistent sub-agents
Tools — how tools are registered and dispatched
Agent — the Agent class and its parameters
Compaction — the turn boundary that message injection reuses