hooks

Bases: Protocol

Duck-typed contract the display hooks require of a display-state object.

The concrete state class lives in the consumer/display layer (a separate task). Depending on this Protocol rather than the concrete class keeps the framework free of a dependency on its own consumer.

start_team(team_id, mandate, agent_count)

111
112
113

def start_team(self, team_id: str, mandate: str, agent_count: int) -> None:
    """Record that a team was dispatched."""
    ...

complete_team(team_id, *, status)

115
116
117

def complete_team(self, team_id: str, *, status: str) -> None:
    """Record that a dispatched team finished."""
    ...

record_tool(*, tool_id, tool_name, status, result)

119
120
121

def record_tool(self, *, tool_id: str, tool_name: str, status: str, result: str) -> None:
    """Record a completed tool call for display."""
    ...

Bases: RuntimeError

Raised by :class:FailFastHook when a critical tool returns an error.

127
128
129

def __init__(self, tool_name: str, error_message: str) -> None:  # noqa: D107
    self.tool_name = tool_name
    super().__init__(f"Tool '{tool_name}' failed: {error_message}")

Bases: HookProvider

Logs a warning when a tool produces unusually large output.

register_hooks(registry, **kwargs)

135
136
137

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the output-quality callback."""
    registry.add_callback(AfterToolCallEvent, self._check_output_quality)

Bases: HookProvider

Strips large toolUse/toolResult payloads from older conversation history.

After the model has reasoned about tool results, the raw payloads are no longer needed. Before each model invocation this hook replaces large tool payloads in older messages with stubs, keeping all reasoning turns and the most recent messages intact — preventing context overflow without ever truncating data the model has not yet seen.

register_hooks(registry, **kwargs)

160
161
162

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the conversation-compaction callback."""
    registry.add_callback(BeforeInvocationEvent, self._compact_history)

Bases: HookProvider

Warns when shell is used for a task that has a dedicated tool.

The mapping of shell-command substring -> suggested alternative is supplied by the use case (alternatives), so the framework holds no domain literals.

219
220

def __init__(self, alternatives: Mapping[str, str]) -> None:  # noqa: D107
    self._alternatives = alternatives

register_hooks(registry, **kwargs)

222
223
224

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the tool-efficiency callback."""
    registry.add_callback(BeforeToolCallEvent, self._check_tool_efficiency)

Bases: HookProvider

Appends a reasoning-checkpoint prompt to results of mapped analysis tools.

Only tools present in the prompts map (supplied by the use case) get a checkpoint appended; every other tool's result passes through unchanged (AC-3-2). Forces the model to interpret combined signals rather than merely collecting tool outputs.

252
253

def __init__(self, prompts: Mapping[str, str]) -> None:  # noqa: D107
    self._prompts = prompts

register_hooks(registry, **kwargs)

255
256
257

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the reasoning-checkpoint callback."""
    registry.add_callback(AfterToolCallEvent, self._inject_reasoning_prompt)

Bases: HookProvider

Raises :class:FullModeToolError when a non-exempt tool errors (full mode).

In full mode, most tool failures should halt the run rather than silently degrade. The exempt tool-name set and exempt name-prefixes both come from the use case (search tools may return no results; external MCP tools may be unavailable), so the framework holds no domain literals (AC-3-4).

287
288
289

def __init__(self, exempt: frozenset[str], exempt_prefixes: tuple[str, ...] = ()) -> None:  # noqa: D107
    self._exempt = exempt
    self._exempt_prefixes = exempt_prefixes

register_hooks(registry, **kwargs)

291
292
293

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the fail-fast callback."""
    registry.add_callback(AfterToolCallEvent, self._check_for_error)

Bases: HookProvider

Triggers revision passes on weak rubric scores OR uncovered acceptance criteria.

After the coordinator finishes an invocation this hook applies two gates (AC-4-1):

1. Rubric. Scores every artifact under layout.artifacts_dir() (matched by layout.artifact_glob()) with the shared rubric.score. An artifact below the rubric's revision threshold is "weak".
2. Acceptance-criteria coverage. Reads every dispatched team's acceptance_criteria from its contract.json (written by dispatch_team) and checks each criterion against the corpus of produced artifacts + team findings. A criterion whose significant tokens are mostly absent from that corpus is "uncovered".

If any artifact is weak OR any criterion is uncovered, the run needs revision and event.resume (verified writable on AfterInvocationEvent in strands 1.47.0) is set to re-invoke the coordinator. An uncovered criterion forces a revision regardless of the rubric total (AC-4-2) and the emitted resume prompt names the uncovered criteria (AC-4-3). When all criteria are covered and no artifact is weak, no resume is emitted (AC-4-4). When no acceptance criteria were supplied for the run, the hook falls back to unchanged rubric-only behavior (AC-4-5). The single max_passes budget caps both the rubric- and coverage-driven resumes combined (AC-4-6).

342
343
344
345
346
347
348
349
350
351
352
353
354
355

def __init__(  # noqa: D107
    self,
    layout: TeamsLayout,
    rubric: NarrativeRubric,
    max_passes: int = 2,
    resume_template: str = DEFAULT_RESUME_TEMPLATE,
    coverage_resume_template: str = DEFAULT_COVERAGE_RESUME_TEMPLATE,
) -> None:
    self._layout = layout
    self._rubric = rubric
    self._max_passes = max_passes
    self._resume_template = resume_template
    self._coverage_resume_template = coverage_resume_template
    self._pass_count = 0

register_hooks(registry, **kwargs)

357
358
359

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the artifact-quality callback."""
    registry.add_callback(AfterInvocationEvent, self._check_artifact_quality)

Bases: HookProvider

Tracks tool calls on a display state for TUI rendering.

Depends on the duck-typed :class:TeamDisplayState Protocol rather than a concrete state class so the framework does not import its own consumer.

503
504

def __init__(self, state: TeamDisplayState) -> None:  # noqa: D107
    self._state = state

register_hooks(registry, **kwargs)

506
507
508
509

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the tool-display callbacks."""
    registry.add_callback(BeforeToolCallEvent, self._on_tool_start)
    registry.add_callback(AfterToolCallEvent, self._on_tool_end)

Bases: HookProvider

Tracks dispatch_team tool calls on a display state.

dispatch_team is a framework-level core tool (not a domain literal), so the name is fixed. Depends on the duck-typed :class:TeamDisplayState.

541
542

def __init__(self, state: TeamDisplayState) -> None:  # noqa: D107
    self._state = state

register_hooks(registry, **kwargs)

544
545
546
547

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the team-dispatch callbacks."""
    registry.add_callback(BeforeToolCallEvent, self._on_dispatch_start)
    registry.add_callback(AfterToolCallEvent, self._on_dispatch_end)

Bases: HookProvider

Emits one structured JSON log line per tool start/end for --quiet mode.

579
580

def __init__(self) -> None:  # noqa: D107
    self._json_logger = logger.bind(output="json")

register_hooks(registry, **kwargs)

582
583
584
585

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the JSON-log callbacks."""
    registry.add_callback(BeforeToolCallEvent, self._on_tool_start)
    registry.add_callback(AfterToolCallEvent, self._on_tool_end)

Bases: RuntimeError

Raised by :class:PolicyGateHook to block a denied consequential tool call (AC-6-4).

607
608
609
610
611
612

def __init__(self, tool_name: str, verdict: Any) -> None:  # noqa: D107
    self.tool_name = tool_name
    self.verdict = verdict
    gate = getattr(verdict, "failed_gate", None)
    reason = getattr(verdict, "reason", "")
    super().__init__(f"Policy gate '{gate}' blocked tool '{tool_name}': {reason}")

Bases: HookProvider

Structurally blocks a denied consequential tool call before it runs (AC-6-4, §14.7).

On BeforeToolCallEvent for a tool whose name is in consequential_tools, the hook builds a :class:~strands_teams.models.TransitionRequest from the tool input and the current actor, calls :meth:~strands_teams.policy.PolicyEngine.authorize_transition (reading grants, approvals, findings, and revision hashes from the event's invocation_state by reference — AC-3-7), and raises :class:PolicyGateDenied if the verdict is allowed=False — so the action never executes without a valid revision-bound approval and a clear blocking-finding slate (AC-6-2, AC-6-3). Non-consequential tools pass straight through; the engine is present but, on a run that requests no consequential transitions, is never invoked to deny (AC-7-5).

The transition kind, target revision, and presented tokens are read from the tool input under conventional keys (transition_kind/kind, target_revision_id, presented_tokens, consequential); a use case that names them differently supplies an input_mapper callable.

Wire the gate to a policy engine and the set of consequential tools.

Parameters:

635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655

def __init__(
    self,
    engine: Any,
    consequential_tools: frozenset[str],
    *,
    actor_of: Any = None,
    input_mapper: Any = None,
) -> None:
    """Wire the gate to a policy engine and the set of consequential tools.

    Args:
        engine: The :class:`~strands_teams.policy.PolicyEngine` that issues verdicts.
        consequential_tools: Tool names whose calls are gated.
        actor_of: Optional callable mapping ``invocation_state`` to the actor id.
        input_mapper: Optional callable ``(tool_name, args, invocation_state)`` ->
            ``TransitionRequest`` for use cases that name inputs differently.
    """
    self._engine = engine
    self._consequential = consequential_tools
    self._actor_of = actor_of
    self._input_mapper = input_mapper

register_hooks(registry, **kwargs)

657
658
659

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the pre-tool authorization callback."""
    registry.add_callback(BeforeToolCallEvent, self._authorize)