WorkflowHooks

Lifecycle callbacks for workflow observability.

Quick Example

from mamba_agents import WorkflowHooks

def log_step(state, step):
    print(f"Step {step.step_number}: {step.description}")

hooks = WorkflowHooks(
    on_workflow_start=lambda s: print("Starting..."),
    on_workflow_complete=lambda r: print(f"Done: {r.success}"),
    on_workflow_error=lambda s, e: print(f"Error: {e}"),
    on_step_start=lambda s, n, t: print(f"Step {n}"),
    on_step_complete=log_step,
    on_step_error=lambda s, step, e: print(f"Step failed: {e}"),
    on_iteration_start=lambda s, i: print(f"Iteration {i}"),
    on_iteration_complete=lambda s, i: print(f"Iteration {i} done"),
)

Available Hooks

Hook	Parameters	Description
on_workflow_start	(state)	Workflow begins
on_workflow_complete	(result)	Workflow completes
on_workflow_error	(state, error)	Workflow fails
on_step_start	(state, step_num, step_type)	Step starts
on_step_complete	(state, step)	Step completes
on_step_error	(state, step, error)	Step fails
on_iteration_start	(state, iteration)	Iteration starts
on_iteration_complete	(state, iteration)	Iteration ends

API Reference

WorkflowHooks

WorkflowHooks(
    on_workflow_start: WorkflowStartHook[StateT]
    | None = None,
    on_workflow_complete: WorkflowCompleteHook[
        OutputT, StateT
    ]
    | None = None,
    on_workflow_error: WorkflowErrorHook[StateT]
    | None = None,
    on_step_start: StepStartHook[StateT] | None = None,
    on_step_complete: StepCompleteHook[StateT]
    | None = None,
    on_step_error: StepErrorHook[StateT] | None = None,
    on_iteration_start: IterationStartHook[StateT]
    | None = None,
    on_iteration_complete: IterationCompleteHook[StateT]
    | None = None,
)

Hook callbacks for workflow execution events.

All hooks are optional and receive workflow state/context. Hooks can be sync or async functions.

ATTRIBUTE	DESCRIPTION
on_workflow_start	Called when workflow begins.
on_workflow_complete	Called when workflow completes successfully.
on_workflow_error	Called when workflow fails.
on_step_start	Called before each step executes.
on_step_complete	Called after each step completes.
on_step_error	Called when a step fails.
on_iteration_start	Called at the start of each iteration.
on_iteration_complete	Called at the end of each iteration.

Initialize workflow hooks.

PARAMETER	DESCRIPTION
on_workflow_start	Called when workflow begins. TYPE: WorkflowStartHook[StateT] | None DEFAULT: None
on_workflow_complete	Called when workflow completes successfully. TYPE: WorkflowCompleteHook[OutputT, StateT] | None DEFAULT: None
on_workflow_error	Called when workflow fails. TYPE: WorkflowErrorHook[StateT] | None DEFAULT: None
on_step_start	Called before each step executes. TYPE: StepStartHook[StateT] | None DEFAULT: None
on_step_complete	Called after each step completes. TYPE: StepCompleteHook[StateT] | None DEFAULT: None
on_step_error	Called when a step fails. TYPE: StepErrorHook[StateT] | None DEFAULT: None
on_iteration_start	Called at the start of each iteration. TYPE: IterationStartHook[StateT] | None DEFAULT: None
on_iteration_complete	Called at the end of each iteration. TYPE: IterationCompleteHook[StateT] | None DEFAULT: None

Source code in src/mamba_agents/workflows/hooks.py

def __init__(
    self,
    on_workflow_start: WorkflowStartHook[StateT] | None = None,
    on_workflow_complete: WorkflowCompleteHook[OutputT, StateT] | None = None,
    on_workflow_error: WorkflowErrorHook[StateT] | None = None,
    on_step_start: StepStartHook[StateT] | None = None,
    on_step_complete: StepCompleteHook[StateT] | None = None,
    on_step_error: StepErrorHook[StateT] | None = None,
    on_iteration_start: IterationStartHook[StateT] | None = None,
    on_iteration_complete: IterationCompleteHook[StateT] | None = None,
) -> None:
    """Initialize workflow hooks.

    Args:
        on_workflow_start: Called when workflow begins.
        on_workflow_complete: Called when workflow completes successfully.
        on_workflow_error: Called when workflow fails.
        on_step_start: Called before each step executes.
        on_step_complete: Called after each step completes.
        on_step_error: Called when a step fails.
        on_iteration_start: Called at the start of each iteration.
        on_iteration_complete: Called at the end of each iteration.
    """
    self.on_workflow_start = on_workflow_start
    self.on_workflow_complete = on_workflow_complete
    self.on_workflow_error = on_workflow_error
    self.on_step_start = on_step_start
    self.on_step_complete = on_step_complete
    self.on_step_error = on_step_error
    self.on_iteration_start = on_iteration_start
    self.on_iteration_complete = on_iteration_complete

trigger_workflow_start async

trigger_workflow_start(
    state: WorkflowState[StateT],
) -> None

Trigger workflow start hook.

PARAMETER	DESCRIPTION
state	Current workflow state. TYPE: WorkflowState[StateT]

Source code in src/mamba_agents/workflows/hooks.py

async def trigger_workflow_start(self, state: WorkflowState[StateT]) -> None:
    """Trigger workflow start hook.

    Args:
        state: Current workflow state.
    """
    await self._trigger("on_workflow_start", state)

trigger_workflow_complete async

trigger_workflow_complete(
    result: WorkflowResult[OutputT, StateT],
) -> None

Trigger workflow complete hook.

PARAMETER	DESCRIPTION
result	Workflow result. TYPE: WorkflowResult[OutputT, StateT]

Source code in src/mamba_agents/workflows/hooks.py

async def trigger_workflow_complete(
    self,
    result: WorkflowResult[OutputT, StateT],
) -> None:
    """Trigger workflow complete hook.

    Args:
        result: Workflow result.
    """
    await self._trigger("on_workflow_complete", result)

trigger_workflow_error async

trigger_workflow_error(
    state: WorkflowState[StateT], error: Exception
) -> None

Trigger workflow error hook.

PARAMETER	DESCRIPTION
state	Current workflow state. TYPE: WorkflowState[StateT]
error	The exception that caused the failure. TYPE: Exception

Source code in src/mamba_agents/workflows/hooks.py

async def trigger_workflow_error(
    self,
    state: WorkflowState[StateT],
    error: Exception,
) -> None:
    """Trigger workflow error hook.

    Args:
        state: Current workflow state.
        error: The exception that caused the failure.
    """
    await self._trigger("on_workflow_error", state, error)

trigger_step_start async

trigger_step_start(
    state: WorkflowState[StateT],
    step_number: int,
    step_type: str,
) -> None

Trigger step start hook.

PARAMETER	DESCRIPTION
state	Current workflow state. TYPE: WorkflowState[StateT]
step_number	The step number about to execute. TYPE: int
step_type	Type identifier of the step. TYPE: str

Source code in src/mamba_agents/workflows/hooks.py

async def trigger_step_start(
    self,
    state: WorkflowState[StateT],
    step_number: int,
    step_type: str,
) -> None:
    """Trigger step start hook.

    Args:
        state: Current workflow state.
        step_number: The step number about to execute.
        step_type: Type identifier of the step.
    """
    await self._trigger("on_step_start", state, step_number, step_type)

trigger_step_complete async

trigger_step_complete(
    state: WorkflowState[StateT], step: WorkflowStep[Any]
) -> None

Trigger step complete hook.

PARAMETER	DESCRIPTION
state	Current workflow state. TYPE: WorkflowState[StateT]
step	The completed step. TYPE: WorkflowStep[Any]

Source code in src/mamba_agents/workflows/hooks.py

async def trigger_step_complete(
    self,
    state: WorkflowState[StateT],
    step: WorkflowStep[Any],
) -> None:
    """Trigger step complete hook.

    Args:
        state: Current workflow state.
        step: The completed step.
    """
    await self._trigger("on_step_complete", state, step)

trigger_step_error async

trigger_step_error(
    state: WorkflowState[StateT],
    step: WorkflowStep[Any],
    error: Exception,
) -> None

Trigger step error hook.

PARAMETER	DESCRIPTION
state	Current workflow state. TYPE: WorkflowState[StateT]
step	The failed step. TYPE: WorkflowStep[Any]
error	The exception that caused the failure. TYPE: Exception

Source code in src/mamba_agents/workflows/hooks.py

async def trigger_step_error(
    self,
    state: WorkflowState[StateT],
    step: WorkflowStep[Any],
    error: Exception,
) -> None:
    """Trigger step error hook.

    Args:
        state: Current workflow state.
        step: The failed step.
        error: The exception that caused the failure.
    """
    await self._trigger("on_step_error", state, step, error)

trigger_iteration_start async

trigger_iteration_start(
    state: WorkflowState[StateT], iteration: int
) -> None

Trigger iteration start hook.

PARAMETER	DESCRIPTION
state	Current workflow state. TYPE: WorkflowState[StateT]
iteration	The iteration number starting. TYPE: int

Source code in src/mamba_agents/workflows/hooks.py

async def trigger_iteration_start(
    self,
    state: WorkflowState[StateT],
    iteration: int,
) -> None:
    """Trigger iteration start hook.

    Args:
        state: Current workflow state.
        iteration: The iteration number starting.
    """
    await self._trigger("on_iteration_start", state, iteration)

trigger_iteration_complete async

trigger_iteration_complete(
    state: WorkflowState[StateT], iteration: int
) -> None

Trigger iteration complete hook.

PARAMETER	DESCRIPTION
state	Current workflow state. TYPE: WorkflowState[StateT]
iteration	The iteration number completed. TYPE: int

Source code in src/mamba_agents/workflows/hooks.py

async def trigger_iteration_complete(
    self,
    state: WorkflowState[StateT],
    iteration: int,
) -> None:
    """Trigger iteration complete hook.

    Args:
        state: Current workflow state.
        iteration: The iteration number completed.
    """
    await self._trigger("on_iteration_complete", state, iteration)