Playbooks

Structured execution plans for agents — declared step sequences with expected tools, validation criteria, and guidance hints. When attached via AgentConfig.playbook, the PlaybookEnforcerHook (the HookProvider the agent installs automatically) drives a PlaybookEnforcer engine that gates each tool call against the current step and auto-advances when the step's expected_tools are exhausted.

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

StepExecution

Bases: BaseModel

Record of a single step's execution.

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

from_playbook classmethod

from_playbook(playbook: Playbook, block_violations: bool = True, record_violations: bool = True) -> 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

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

    Returns:
        Configured PlaybookEnforcer
    """
    plan = PlaybookPlan(playbook=playbook)
    return cls(
        plan=plan,
        block_violations=block_violations,
        record_violations=record_violations,
    )

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)

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

Record that a tool was called.

Updates the step execution tracking.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool that was called	required

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

def record_tool_call(self, tool_name: str) -> None:
    """Record that a tool was called.

    Updates the step execution tracking.

    Args:
        tool_name: Name of the tool that was called
    """
    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

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]
    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.
        self._enforcer.record_tool_call(event.tool_name)
        return

    self._enforcer.record_tool_call(event.tool_name)

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