Playbooks

Structured execution plans for agents — declared step sequences with expected tools, validation criteria, and guidance hints. Attach one via AgentConfig.playbook; enforcement details below.

Models

Playbook

Bases: BaseModel

Collection of steps that define an execution plan.

A playbook provides structure for agent execution, defining the expected sequence of operations and validation criteria.

get_step

get_step(step_id: str) -> PlaybookStep | None

Get a step by its ID.

Source code in .sdk/src/tulip/playbooks/models.py

def get_step(self, step_id: str) -> PlaybookStep | None:
    """Get a step by its ID."""
    for step in self.steps:
        if step.id == step_id:
            return step
    return None

get_step_index

get_step_index(step_id: str) -> int | None

Get the index of a step by its ID.

Source code in .sdk/src/tulip/playbooks/models.py

def get_step_index(self, step_id: str) -> int | None:
    """Get the index of a step by its ID."""
    for i, step in enumerate(self.steps):
        if step.id == step_id:
            return i
    return None

PlaybookStep

Bases: BaseModel

Individual step in a playbook.

Defines what tools are expected, hints for the agent, and optional validation criteria.

PlaybookPlan

Bases: BaseModel

Active execution plan for a playbook.

Tracks progress through the playbook, including which steps have been completed, current step, and any deviations.

current_step property

current_step: PlaybookStep | None

Get the current step.

progress property

progress: float

Calculate progress as a percentage (0.0 to 1.0).

completed_steps property

completed_steps: list[str]

Get IDs of completed steps.

pending_steps property

pending_steps: list[str]

Get IDs of pending steps.

get_step_execution

get_step_execution(step_id: str) -> StepExecution | None

Get execution record for a step.

Source code in .sdk/src/tulip/playbooks/models.py

def get_step_execution(self, step_id: str) -> StepExecution | None:
    """Get execution record for a step."""
    return self.step_executions.get(step_id)

is_step_complete

is_step_complete(step_id: str) -> bool

Check if a step is complete.

Source code in .sdk/src/tulip/playbooks/models.py

def is_step_complete(self, step_id: str) -> bool:
    """Check if a step is complete."""
    se = self.step_executions.get(step_id)
    return se is not None and se.status == StepStatus.COMPLETED

unresolved_required_steps

unresolved_required_steps() -> list[str]

Required steps that never completed — the conclusion contract.

A run may not honestly conclude while one of these is outstanding.

Source code in .sdk/src/tulip/playbooks/models.py

def unresolved_required_steps(self) -> list[str]:
    """Required steps that never completed — the conclusion contract.

    A run may not honestly conclude while one of these is outstanding.
    """
    done = {
        step_id
        for step_id, execution in self.step_executions.items()
        if execution.status == StepStatus.COMPLETED
    }
    return [step.id for step in self.playbook.steps if step.required and step.id not in done]

StepExecution

Bases: BaseModel

Record of a single step's execution.

probe_coverage

probe_coverage(probes: list[RequiredProbe]) -> float

matched / required, and 1.0 when nothing was required.

Takes the probes rather than the step: the authoritative set is a step's OWN probes plus those of every skill it uses, and only the enforcer can resolve skills. Reading them off the step here produced a run that reported 1.00 adherence while a violation on the same step said 1 of 3 matched — two answers to one question, found by running a real scenario rather than a fixture.

A step that asked for nothing is fully covered by definition — the alternative is that every legacy playbook reports zero adherence, which would make the number worthless on the day it shipped.

Source code in .sdk/src/tulip/playbooks/models.py

def probe_coverage(self, probes: list[RequiredProbe]) -> float:
    """matched / required, and 1.0 when nothing was required.

    Takes the probes rather than the step: the authoritative set is a step's
    OWN probes plus those of every skill it `uses`, and only the enforcer
    can resolve skills. Reading them off the step here produced a run that
    reported 1.00 adherence while a violation on the same step said 1 of 3
    matched — two answers to one question, found by running a real scenario
    rather than a fixture.

    A step that asked for nothing is fully covered by definition — the
    alternative is that every legacy playbook reports zero adherence, which
    would make the number worthless on the day it shipped.
    """
    required = {probe.name for probe in probes}
    if not required:
        return 1.0
    return len(required & set(self.matched_probes)) / len(required)

unmatched_probes

unmatched_probes(probes: list[RequiredProbe]) -> list[str]

What this step was told to look at and did not — in declared order.

Source code in .sdk/src/tulip/playbooks/models.py

def unmatched_probes(self, probes: list[RequiredProbe]) -> list[str]:
    """What this step was told to look at and did not — in declared order."""
    matched = set(self.matched_probes)
    return [probe.name for probe in probes if probe.name not in matched]

StepStatus

Bases: StrEnum

Status of a playbook step.

Loader

load_playbook

load_playbook(source: str | Path | dict[str, Any]) -> Playbook

Load a playbook from various sources.

Parameters:

Name	Type	Description	Default
source	str | Path | dict[str, Any]	Path to file, JSON string, or dictionary	required

Returns:

Type	Description
Playbook	Loaded and validated Playbook

Examples:

>>> playbook = load_playbook("./playbooks/deploy.yaml")
>>> playbook = load_playbook({"id": "test", "name": "Test", "steps": []})

Source code in .sdk/src/tulip/playbooks/loader.py

def load_playbook(source: str | Path | dict[str, Any]) -> Playbook:
    """Load a playbook from various sources.

    Args:
        source: Path to file, JSON string, or dictionary

    Returns:
        Loaded and validated Playbook

    Examples:
        >>> playbook = load_playbook("./playbooks/deploy.yaml")
        >>> playbook = load_playbook({"id": "test", "name": "Test", "steps": []})
    """
    loader = PlaybookLoader()

    if isinstance(source, dict):
        return loader.load_dict(source)

    if isinstance(source, Path):
        return loader.load_file(source)

    # String - could be path or JSON
    source_str = str(source)

    # Check if it's a file path
    path = Path(source_str)
    if path.exists():
        return loader.load_file(path)

    # Try as JSON string
    if source_str.strip().startswith("{"):
        return loader.load_json_string(source_str)

    # Assume it's a non-existent file path
    raise PlaybookLoadError(f"File not found: {source_str}", path=path)

PlaybookLoader

Load playbooks from JSON and YAML files.

Supports loading from: - JSON files (.json) - YAML files (.yaml, .yml) - Dictionaries (for programmatic use)

load_file

load_file(path: str | Path) -> Playbook

Load a playbook from a file.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the playbook file (.json, .yaml, or .yml)	required

Returns:

Type	Description
Playbook	Loaded and validated Playbook

Raises:

Type	Description
PlaybookLoadError	If file cannot be loaded or validated

Source code in .sdk/src/tulip/playbooks/loader.py

def load_file(self, path: str | Path) -> Playbook:
    """Load a playbook from a file.

    Args:
        path: Path to the playbook file (.json, .yaml, or .yml)

    Returns:
        Loaded and validated Playbook

    Raises:
        PlaybookLoadError: If file cannot be loaded or validated
    """
    path = Path(path)

    if not path.exists():
        raise PlaybookLoadError(f"File not found: {path}", path=path)

    suffix = path.suffix.lower()

    try:
        if suffix == ".json":
            return self._load_json(path)
        if suffix in (".yaml", ".yml"):
            return self._load_yaml(path)
        raise PlaybookLoadError(
            f"Unsupported file format: {suffix}. Use .json, .yaml, or .yml",
            path=path,
        )
    except PlaybookLoadError:
        raise
    except Exception as e:
        raise PlaybookLoadError(f"Failed to load {path}: {e}", path=path) from e

load_dict

load_dict(data: dict[str, Any]) -> Playbook

Load a playbook from a dictionary.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary containing playbook definition	required

Returns:

Type	Description
Playbook	Loaded and validated Playbook

Raises:

Type	Description
PlaybookLoadError	If data is invalid

Source code in .sdk/src/tulip/playbooks/loader.py

def load_dict(self, data: dict[str, Any]) -> Playbook:
    """Load a playbook from a dictionary.

    Args:
        data: Dictionary containing playbook definition

    Returns:
        Loaded and validated Playbook

    Raises:
        PlaybookLoadError: If data is invalid
    """
    data = _flatten_step_groups(data)
    errors = self._validate_structure(data)
    if errors:
        raise PlaybookLoadError(
            f"Invalid playbook structure: {len(errors)} errors",
            errors=errors,
        )

    try:
        return Playbook(**data)
    except ValidationError as e:
        errors = [str(err) for err in e.errors()]
        raise PlaybookLoadError(
            f"Playbook validation failed: {len(errors)} errors",
            errors=errors,
        ) from e

load_json_string

load_json_string(json_string: str) -> Playbook

Load a playbook from a JSON string.

Parameters:

Name	Type	Description	Default
json_string	str	JSON string containing playbook definition	required

Returns:

Type	Description
Playbook	Loaded and validated Playbook

Raises:

Type	Description
PlaybookLoadError	If JSON is invalid or playbook validation fails

Source code in .sdk/src/tulip/playbooks/loader.py

def load_json_string(self, json_string: str) -> Playbook:
    """Load a playbook from a JSON string.

    Args:
        json_string: JSON string containing playbook definition

    Returns:
        Loaded and validated Playbook

    Raises:
        PlaybookLoadError: If JSON is invalid or playbook validation fails
    """
    try:
        data = json.loads(json_string)
    except json.JSONDecodeError as e:
        raise PlaybookLoadError(f"Invalid JSON: {e}") from e

    return self.load_dict(data)

load_yaml_string

load_yaml_string(yaml_string: str) -> Playbook

Load a playbook from a YAML string.

Parameters:

Name	Type	Description	Default
yaml_string	str	YAML string containing playbook definition	required

Returns:

Type	Description
Playbook	Loaded and validated Playbook

Raises:

Type	Description
PlaybookLoadError	If YAML is invalid or playbook validation fails

Source code in .sdk/src/tulip/playbooks/loader.py

def load_yaml_string(self, yaml_string: str) -> Playbook:
    """Load a playbook from a YAML string.

    Args:
        yaml_string: YAML string containing playbook definition

    Returns:
        Loaded and validated Playbook

    Raises:
        PlaybookLoadError: If YAML is invalid or playbook validation fails
    """
    try:
        import yaml  # type: ignore[import-untyped]  # PyYAML ships no inline types
    except ImportError as e:
        raise PlaybookLoadError(
            "PyYAML is required for YAML support. Install with: pip install pyyaml"
        ) from e

    try:
        data = yaml.safe_load(yaml_string)
    except yaml.YAMLError as e:
        raise PlaybookLoadError(f"Invalid YAML: {e}") from e

    return self.load_dict(data)

PlaybookLoadError

PlaybookLoadError(message: str, path: Path | None = None, errors: list[str] | None = None)

Bases: Exception

Error loading a playbook.

Source code in .sdk/src/tulip/playbooks/loader.py

def __init__(self, message: str, path: Path | None = None, errors: list[str] | None = None):
    self.path = path
    self.errors = errors or []
    super().__init__(message)

Enforcer

PlaybookEnforcer is the enforcement engine that holds the model to the playbook's step sequence. PlaybookEnforcerHook is the HookProvider wrapper around it, installed automatically when AgentConfig.playbook is set.

PlaybookEnforcer

Bases: BaseModel

Enforces playbook execution sequence and constraints.

The enforcer tracks progress through a playbook, validates tool calls, and provides hints to guide the agent through the execution plan.

Features: - Track completed steps - Validate tool calls match current step's expected tools - Provide hints for the next step - Block out-of-sequence execution when strict_sequence is True - Record violations for auditing

violations property

violations: list[EnforcementViolation]

Get recorded violations.

current_step property

current_step: PlaybookStep | None

Get the current step.

current_step_hints property

current_step_hints: list[str]

Get hints for the current step.

progress property

progress: float

Get execution progress (0.0 to 1.0).

is_complete property

is_complete: bool

Check if the playbook execution is complete.

effective_probes

effective_probes(step: PlaybookStep) -> list[Any]

The step's own probes plus those of every skill it is carried out with.

De-duplicated by name, step-declared first: a step may sharpen a skill's generic probe for its own context, and the more specific declaration is the one an author expects to win.

Source code in .sdk/src/tulip/playbooks/enforcer.py

def effective_probes(self, step: PlaybookStep) -> list[Any]:
    """The step's own probes plus those of every skill it is carried out with.

    De-duplicated by name, step-declared first: a step may sharpen a skill's
    generic probe for its own context, and the more specific declaration is
    the one an author expects to win.
    """
    seen: dict[str, Any] = {}
    for probe in step.required_probes:
        seen.setdefault(probe.name, probe)
    for skill in self._used_skills(step):
        for probe in getattr(skill, "required_probes", ()) or ():
            seen.setdefault(probe.name, probe)
    return list(seen.values())

adherence_score

adherence_score() -> float

How much of the declared evidence this run gathered, 0..1.

Lives HERE, not on the plan, because the evidence a step owes is its own probes plus those of every skill it uses, and only the enforcer holds the skills. The plan-level version counted step-declared probes alone and reported 1.00 for a run whose violations said 1 of 3 matched.

Averaged over steps REACHED, not the whole playbook: a run still in flight should not read as non-compliant for being unfinished, and a run that stopped early is already reported by unresolved_required_steps.

Treat it as a floor, not a grade — see ASSESSMENT-required-probes-critique.md. The violation list is the artefact worth showing a person.

Source code in .sdk/src/tulip/playbooks/enforcer.py

def adherence_score(self) -> float:
    """How much of the declared evidence this run gathered, 0..1.

    Lives HERE, not on the plan, because the evidence a step owes is its own
    probes plus those of every skill it `uses`, and only the enforcer holds
    the skills. The plan-level version counted step-declared probes alone
    and reported 1.00 for a run whose violations said 1 of 3 matched.

    Averaged over steps REACHED, not the whole playbook: a run still in
    flight should not read as non-compliant for being unfinished, and a run
    that stopped early is already reported by `unresolved_required_steps`.

    Treat it as a floor, not a grade — see
    ASSESSMENT-required-probes-critique.md. The violation list is the
    artefact worth showing a person.
    """
    steps = {step.id: step for step in self.plan.playbook.steps}
    covered = [
        execution.probe_coverage(self.effective_probes(steps[step_id]))
        for step_id, execution in self.plan.step_executions.items()
        if step_id in steps
    ]
    return sum(covered) / len(covered) if covered else 1.0

allowed_tools_for

allowed_tools_for(step: PlaybookStep) -> set[str] | None

What this step may call, or None when it does not constrain calls.

THE reason uses had to exist before a skill's allowed_tools could mean anything: a skill is prose folded into a system prompt, so on its own there is no moment at which its allow-list applies. A step supplies that moment. Outside a step that names it, a skill still constrains nothing — which is honest, and is why this returns None rather than an empty set when nothing is declared.

Source code in .sdk/src/tulip/playbooks/enforcer.py

def allowed_tools_for(self, step: PlaybookStep) -> set[str] | None:
    """What this step may call, or None when it does not constrain calls.

    THE reason `uses` had to exist before a skill's `allowed_tools` could
    mean anything: a skill is prose folded into a system prompt, so on its
    own there is no moment at which its allow-list applies. A step supplies
    that moment. Outside a step that names it, a skill still constrains
    nothing — which is honest, and is why this returns None rather than an
    empty set when nothing is declared.
    """
    allowed: set[str] = set()
    declared = False
    for skill in self._used_skills(step):
        tools = getattr(skill, "allowed_tools", None)
        if tools:
            declared = True
            allowed.update(str(tool) for tool in tools)
    return allowed if declared else None

from_playbook classmethod

from_playbook(playbook: Playbook, block_violations: bool = True, record_violations: bool = True, skills: Mapping[str, Any] | None = None) -> PlaybookEnforcer

Create an enforcer from a playbook.

Parameters:

Name	Type	Description	Default
playbook	Playbook	The playbook to enforce	required
block_violations	bool	Whether to block violating tool calls	True
record_violations	bool	Whether to record violations	True
skills	Mapping[str, Any] | None	name -> Skill, for steps that name capabilities via uses. Omitted, those references resolve to nothing and constrain nothing — a playbook may legitimately be enforced somewhere the skill bodies are not loaded, and failing there would make the enforcer refuse to run rather than enforce less.	None

Returns:

Type	Description
PlaybookEnforcer	Configured PlaybookEnforcer

Source code in .sdk/src/tulip/playbooks/enforcer.py

@classmethod
def from_playbook(
    cls,
    playbook: Playbook,
    block_violations: bool = True,
    record_violations: bool = True,
    skills: Mapping[str, Any] | None = None,
) -> PlaybookEnforcer:
    """Create an enforcer from a playbook.

    Args:
        playbook: The playbook to enforce
        block_violations: Whether to block violating tool calls
        record_violations: Whether to record violations
        skills: name -> Skill, for steps that name capabilities via `uses`.
            Omitted, those references resolve to nothing and constrain
            nothing — a playbook may legitimately be enforced somewhere the
            skill bodies are not loaded, and failing there would make the
            enforcer refuse to run rather than enforce less.

    Returns:
        Configured PlaybookEnforcer
    """
    plan = PlaybookPlan(playbook=playbook)
    enforcer = cls(
        plan=plan,
        block_violations=block_violations,
        record_violations=record_violations,
    )
    enforcer._skills = dict(skills or {})
    return enforcer

validate_tool_call

validate_tool_call(tool_name: str) -> EnforcementResult

Validate a tool call against the current step.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool being called	required

Returns:

Type	Description
EnforcementResult	EnforcementResult indicating whether the call is allowed

Source code in .sdk/src/tulip/playbooks/enforcer.py

def validate_tool_call(self, tool_name: str) -> EnforcementResult:
    """Validate a tool call against the current step.

    Args:
        tool_name: Name of the tool being called

    Returns:
        EnforcementResult indicating whether the call is allowed
    """
    step = self.current_step

    # No more steps - check if extra tools are allowed
    if step is None:
        if self.plan.completed:
            return EnforcementResult(
                allowed=self.plan.playbook.allow_extra_tools,
                violation=self._maybe_record_violation(
                    "playbook_complete",
                    None,
                    tool_name,
                    f"Playbook is complete, tool '{tool_name}' called after completion",
                    blocked=self.block_violations and not self.plan.playbook.allow_extra_tools,
                )
                if not self.plan.playbook.allow_extra_tools
                else None,
                hints=["Playbook execution is complete"],
            )
        return EnforcementResult(allowed=True)

    # A skill's allow-list, enforced — inside the step that names it.
    # Until `uses` existed there was no moment at which this could apply,
    # so `allowed_tools` was parsed, printed into the prompt as a sentence,
    # and gated by nothing. A step supplies the scope.
    skill_allowed = self.allowed_tools_for(step)
    if skill_allowed is not None and tool_name not in skill_allowed:
        if self.plan.playbook.allow_extra_tools:
            return EnforcementResult(allowed=True, current_step=step)
        violation = self._maybe_record_violation(
            violation_type="tool_outside_skill",
            step_id=step.id,
            tool_name=tool_name,
            message=(
                f"{tool_name!r} is not among the tools the skill(s) "
                f"{', '.join(step.uses)} allow for step {step.id!r}: "
                f"{', '.join(sorted(skill_allowed))}"
            ),
            blocked=self.block_violations,
        )
        return EnforcementResult(
            allowed=not self.block_violations,
            violation=violation,
            current_step=step,
            hints=list(step.hints),
        )

    # Check if tool is in expected tools
    if step.expected_tools and tool_name not in step.expected_tools:
        # Tool not expected for this step
        if self.plan.playbook.allow_extra_tools:
            return EnforcementResult(
                allowed=True,
                hints=step.hints,
                current_step=step,
            )

        violation = self._maybe_record_violation(
            "unexpected_tool",
            step.id,
            tool_name,
            f"Tool '{tool_name}' not expected for step '{step.id}'. "
            f"Expected: {step.expected_tools}",
            blocked=self.block_violations,
        )

        return EnforcementResult(
            allowed=not self.block_violations,
            violation=violation,
            hints=[
                f"Current step expects: {', '.join(step.expected_tools)}",
                *step.hints,
            ],
            current_step=step,
        )

    # Check max tool calls for step
    step_exec = self.plan.step_executions.get(step.id)
    if step.max_tool_calls is not None and step_exec:
        if step_exec.tool_call_count >= step.max_tool_calls:
            violation = self._maybe_record_violation(
                "max_tool_calls",
                step.id,
                tool_name,
                f"Step '{step.id}' has reached max tool calls ({step.max_tool_calls})",
                blocked=self.block_violations,
            )
            return EnforcementResult(
                allowed=not self.block_violations,
                violation=violation,
                hints=["Consider moving to the next step"],
                current_step=step,
            )

    return EnforcementResult(
        allowed=True,
        hints=step.hints,
        current_step=step,
    )

record_tool_call

record_tool_call(tool_name: str, *, arguments: Any = None, result: Any = None) -> None

Record that a tool was called, and what it looked at.

arguments and result are what the step's required_probes are matched against. They are optional so every existing caller keeps working — a caller that passes neither simply gathers no evidence, and a step with probes will report them unmatched, which is the honest answer rather than a silent pass.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool that was called	required
arguments	Any	The call's arguments, searched for probe matches	None
result	Any	The call's result, searched for probe matches	None

Source code in .sdk/src/tulip/playbooks/enforcer.py

def record_tool_call(
    self,
    tool_name: str,
    *,
    arguments: Any = None,
    result: Any = None,
) -> None:
    """Record that a tool was called, and what it looked at.

    `arguments` and `result` are what the step's ``required_probes`` are
    matched against. They are optional so every existing caller keeps
    working — a caller that passes neither simply gathers no evidence, and
    a step with probes will report them unmatched, which is the honest
    answer rather than a silent pass.

    Args:
        tool_name: Name of the tool that was called
        arguments: The call's arguments, searched for probe matches
        result: The call's result, searched for probe matches
    """
    step = self.current_step
    if step is None:
        self.plan.total_tool_calls += 1
        return

    # Get or create step execution
    if step.id not in self.plan.step_executions:
        self.plan.step_executions[step.id] = StepExecution(
            step_id=step.id,
            status=StepStatus.IN_PROGRESS,
            started_at=datetime.now(UTC),
        )

    step_exec = self.plan.step_executions[step.id]
    step_exec.tool_calls.append(tool_name)
    step_exec.tool_call_count += 1
    self.plan.total_tool_calls += 1
    self._match_probes(step, step_exec, tool_name, arguments, result)

complete_current_step

complete_current_step(result: str | None = None) -> bool

Mark the current step as complete and advance.

Parameters:

Name	Type	Description	Default
result	str | None	Optional result to record for the step	None

Returns:

Type	Description
bool	True if advanced to next step, False if playbook is complete

Source code in .sdk/src/tulip/playbooks/enforcer.py

def complete_current_step(self, result: str | None = None) -> bool:
    """Mark the current step as complete and advance.

    Args:
        result: Optional result to record for the step

    Returns:
        True if advanced to next step, False if playbook is complete
    """
    step = self.current_step
    if step is None:
        return False

    # Get or create step execution
    if step.id not in self.plan.step_executions:
        self.plan.step_executions[step.id] = StepExecution(
            step_id=step.id,
            status=StepStatus.COMPLETED,
            started_at=datetime.now(UTC),
        )

    step_exec = self.plan.step_executions[step.id]
    # Evidence first: a step that gathered less than it declared does not
    # get to close quietly. Recorded as a violation like any other, so it
    # reaches the same trace an out-of-shape call does — the point is that
    # "the procedure was followed" stops meaning "the right tools were
    # called" and starts meaning "the required evidence exists".
    probes = self.effective_probes(step)
    matched = set(step_exec.matched_probes)
    unmatched = step_exec.unmatched_probes(probes)
    if unmatched and self.record_violations:
        self._violations.append(
            EnforcementViolation(
                violation_type="evidence_incomplete",
                step_id=step.id,
                message=(
                    f"step {step.id!r} completed without the evidence it requires: "
                    f"{', '.join(unmatched)} "
                    f"({len(matched & {p.name for p in probes})}/{len(probes)} matched)"
                ),
            )
        )
    floor = step.min_tool_calls
    if floor is not None and step_exec.tool_call_count < floor and self.record_violations:
        self._violations.append(
            EnforcementViolation(
                violation_type="insufficient_effort",
                step_id=step.id,
                message=(
                    f"step {step.id!r} completed after {step_exec.tool_call_count} tool "
                    f"call(s); it declares a floor of {floor}"
                ),
            )
        )
    step_exec.status = StepStatus.COMPLETED
    step_exec.completed_at = datetime.now(UTC)
    step_exec.result = result

    # Advance to next step
    self.plan.current_step_index += 1

    # Check if playbook is complete
    if self.plan.current_step_index >= len(self.plan.playbook.steps):
        self.plan.completed = True
        return False

    return True

skip_current_step

skip_current_step(reason: str | None = None) -> bool

Skip the current step.

Only works for non-required steps.

Parameters:

Name	Type	Description	Default
reason	str | None	Optional reason for skipping	None

Returns:

Type	Description
bool	True if step was skipped, False if step is required

Source code in .sdk/src/tulip/playbooks/enforcer.py

def skip_current_step(self, reason: str | None = None) -> bool:
    """Skip the current step.

    Only works for non-required steps.

    Args:
        reason: Optional reason for skipping

    Returns:
        True if step was skipped, False if step is required
    """
    step = self.current_step
    if step is None:
        return False

    if step.required:
        return False

    # Record as skipped
    if step.id not in self.plan.step_executions:
        self.plan.step_executions[step.id] = StepExecution(
            step_id=step.id,
            status=StepStatus.SKIPPED,
        )
    else:
        self.plan.step_executions[step.id].status = StepStatus.SKIPPED

    if reason:
        self.plan.step_executions[step.id].result = reason

    # Advance
    self.plan.current_step_index += 1

    if self.plan.current_step_index >= len(self.plan.playbook.steps):
        self.plan.completed = True
        return True

    return True

fail_current_step

fail_current_step(error: str) -> None

Mark the current step as failed.

Parameters:

Name	Type	Description	Default
error	str	Error message	required

Source code in .sdk/src/tulip/playbooks/enforcer.py

def fail_current_step(self, error: str) -> None:
    """Mark the current step as failed.

    Args:
        error: Error message
    """
    step = self.current_step
    if step is None:
        return

    if step.id not in self.plan.step_executions:
        self.plan.step_executions[step.id] = StepExecution(
            step_id=step.id,
            status=StepStatus.FAILED,
            started_at=datetime.now(UTC),
        )

    step_exec = self.plan.step_executions[step.id]
    step_exec.status = StepStatus.FAILED
    step_exec.completed_at = datetime.now(UTC)
    step_exec.error = error

    self.plan.errors.append(f"Step {step.id}: {error}")

get_next_step_hints

get_next_step_hints() -> list[str]

Get hints for the next step after current.

Useful for looking ahead during execution.

Returns:

Type	Description
list[str]	List of hints for the next step, or empty if no next step

Source code in .sdk/src/tulip/playbooks/enforcer.py

def get_next_step_hints(self) -> list[str]:
    """Get hints for the next step after current.

    Useful for looking ahead during execution.

    Returns:
        List of hints for the next step, or empty if no next step
    """
    next_index = self.plan.current_step_index + 1
    if next_index < len(self.plan.playbook.steps):
        return list(self.plan.playbook.steps[next_index].hints)
    return []

get_step_summary

get_step_summary() -> dict[str, Any]

Get a summary of step execution status.

Returns:

Type	Description
dict[str, Any]	Dictionary with step status summary

Source code in .sdk/src/tulip/playbooks/enforcer.py

def get_step_summary(self) -> dict[str, Any]:
    """Get a summary of step execution status.

    Returns:
        Dictionary with step status summary
    """
    steps = self.plan.playbook.steps
    return {
        "total_steps": len(steps),
        "current_step_index": self.plan.current_step_index,
        "completed": len(
            [s for s in self.plan.step_executions.values() if s.status == StepStatus.COMPLETED]
        ),
        "skipped": len(
            [s for s in self.plan.step_executions.values() if s.status == StepStatus.SKIPPED]
        ),
        "failed": len(
            [s for s in self.plan.step_executions.values() if s.status == StepStatus.FAILED]
        ),
        "pending": len(steps) - len(self.plan.step_executions),
        "progress": self.progress,
        "is_complete": self.is_complete,
    }

reset

reset() -> None

Reset the enforcer to start over.

Source code in .sdk/src/tulip/playbooks/enforcer.py

def reset(self) -> None:
    """Reset the enforcer to start over."""
    self.plan.current_step_index = 0
    self.plan.step_executions.clear()
    self.plan.completed = False
    self.plan.total_tool_calls = 0
    self.plan.errors.clear()
    self._violations.clear()

EnforcementResult

Bases: BaseModel

Result of an enforcement check.

EnforcementViolation

Bases: BaseModel

Record of an enforcement violation.

PlaybookEnforcerHook

PlaybookEnforcerHook(playbook: Playbook, *, block_violations: bool = True, record_violations: bool = True, priority: int = HookPriority.SECURITY_DEFAULT)

Bases: HookProvider

Hook that enforces a :class:Playbook over an agent run.

Holds a single :class:PlaybookEnforcer instance and dispatches before/after_tool_call events into it so step compliance is tracked automatically. Auto-advances to the next step when the current step's expected tool list is exhausted.

Parameters:

Name	Type	Description	Default
playbook	Playbook	The playbook to enforce. A fresh PlaybookEnforcer is built from it on construction.	required
block_violations	bool	When True (default), a violating tool call is cancelled via event.cancel. When False, violations are recorded but the call still runs.	True
record_violations	bool	When True (default), violations land on enforcer.violations for post-run inspection.	True
priority	int	Hook priority. Defaults to a high value so the enforcer runs before observability / retry hooks; that way a blocked tool call doesn't get logged as if it had executed.	SECURITY_DEFAULT

Example

from tulip import Agent from tulip.playbooks.loader import load_playbook from tulip.playbooks.hook import PlaybookEnforcerHook

playbook = load_playbook("playbooks/triage.yaml") agent = Agent( model="openai:gpt-4o", tools=[search, classify, escalate], hooks=[PlaybookEnforcerHook(playbook)], ) result = agent.run_sync("Triage this incident.")

Source code in .sdk/src/tulip/playbooks/hook.py

def __init__(
    self,
    playbook: Playbook,
    *,
    block_violations: bool = True,
    record_violations: bool = True,
    priority: int = HookPriority.SECURITY_DEFAULT,
) -> None:
    self._enforcer = PlaybookEnforcer.from_playbook(
        playbook,
        block_violations=block_violations,
        record_violations=record_violations,
    )
    self._priority = priority

enforcer property

enforcer: PlaybookEnforcer

Return the underlying enforcer for inspection (violations, progress).

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Validate the call against the current step; cancel on violation.

Source code in .sdk/src/tulip/playbooks/hook.py

async def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
    """Validate the call against the current step; cancel on violation."""
    # ``ProtectedEvent`` (the base class for hook events) sets fields
    # via ``self._init(name, value)`` rather than class-level annotations,
    # so mypy can't see ``.tool_name`` / ``.error`` statically. They
    # exist at runtime; the ignore is the standard pattern for this
    # protocol.
    result = self._enforcer.validate_tool_call(event.tool_name)
    if result.allowed:
        return
    # Build a useful cancel message that the agent loop will turn into
    # a tool result so the model can recover. The hint list is the
    # enforcer's machine-readable "what to do next" for the model.
    msg_parts = []
    if result.violation is not None:
        msg_parts.append(result.violation.message)
    if result.hints:
        msg_parts.append("Hints: " + " ".join(result.hints))
    event.cancel = "PlaybookEnforcer blocked: " + " | ".join(msg_parts)

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Record the call and auto-advance when the current step is satisfied.

The agent loop short-circuits past on_after_tool_call when the before-hook cancelled the call, so anything reaching this method actually executed.

Source code in .sdk/src/tulip/playbooks/hook.py

async def on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    """Record the call and auto-advance when the current step is satisfied.

    The agent loop short-circuits past ``on_after_tool_call`` when the
    before-hook cancelled the call, so anything reaching this method
    actually executed.
    """
    # ``ProtectedEvent`` sets ``.error`` / ``.tool_name`` via
    # ``self._init(...)`` not class-level fields — see note in
    # ``on_before_tool_call``.
    if event.error:
        # Failed calls don't advance the step (the model will likely
        # retry); they're still recorded for the violation log. The
        # ARGUMENTS still count as evidence-seeking — the agent did look —
        # but the result does not, because there isn't one.
        self._enforcer.record_tool_call(
            event.tool_name, arguments=getattr(event, "arguments", None)
        )
        return

    # Arguments AND result, because a step's `required_probes` may be
    # satisfied by either: what was asked for, or what came back. A tool
    # asked a general question can answer a specific one.
    self._enforcer.record_tool_call(
        event.tool_name,
        arguments=getattr(event, "arguments", None),
        result=getattr(event, "result", None),
    )

    step = self._enforcer.current_step
    if step is None:
        return

    step_exec = self._enforcer.plan.step_executions.get(step.id)
    if step_exec is None:
        return

    # Auto-advance the plan when the step's expected tools have all
    # been seen, OR when max_tool_calls is reached. Without this, the
    # enforcer would block legitimate next-step calls because the plan
    # is still pointing at a satisfied step.
    if step.expected_tools:
        seen = set(step_exec.tool_calls)
        if set(step.expected_tools).issubset(seen):
            self._enforcer.complete_current_step()
            return

    if step.max_tool_calls is not None and step_exec.tool_call_count >= step.max_tool_calls:
        self._enforcer.complete_current_step()

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Called before agent starts processing.

Parameters:

Name	Type	Description	Default
prompt	str	The user prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Called before agent starts processing.

    Args:
        prompt: The user prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in .sdk/src/tulip/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }