Playbooks

Structured execution plans for agents — declared step sequences with expected tools, validation criteria, and guidance hints. When attached via AgentConfig.playbook, the PlaybookEnforcer hook 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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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

The hook that holds the model to the playbook's step sequence. 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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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 src/locus/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.