Workflow Notifications

Overview

This section details how to set up and manage webhook notifications for ongoing workflows.

Hooks

Hooks allow users to trigger custom actions based on specific workflow events, such as task completion, state changes, or safety alerts. Hooks are typically used to send HTTP requests (webhooks) to external systems, allowing for real-time monitoring and integration with other services.

To define a webhook, you need to specify the URL and hook type (RunStateChangeHook) for the HTTP request that will be sent when the hook is triggered.

from linq.hooks import (
    HookParameters,
    LabwareMovementHook,
    NewPlanHook,
    RunStateChangeHook,
    SafetyStateChangeHook,
    TaskStateChangeHook,
)

hooks = [
    RunStateChangeHook(
        parameters=HookParameters(
            url="http://example.com",
        )
    ),
    ...
]

workflow = Workflow(
    ...
    hooks=hooks,
)

Types of Hooks

There are several types of hooks available in the LINQ scheduler:

1. RunStateChangeHook: Triggeres a HTTP POST request when the state of a workflow run changes (e.g. start, pause, stop).

hooks = [
    RunStateChangeHook(
        parameters=HookParameters(
            url="http://example.com",
        )
    ),
    ]

class linq.hooks.RunStateChangeHook(*, parameters: HookParameters, filter: Literal[Filters.ON_RUN_STATE_CHANGE] = FieldInfo(annotation=NoneType, required=True, description='Event type triggering the hook. No longer required.', deprecated=True))

Hook to be triggered when a run state changes.

2. TaskStateChangeHook: Triggered the state of a specific task changes (e.g. task success or failure).

   Parameters:

   • task_ids: List of task IDs that should trigger the hook. If left empty, the hook will trigger for all tasks.

hooks = [
    TaskStateChangeHook(
        parameters=HookParameters(
            url="http://example.com",
        ),
        task_ids=[task_id],  # Specify tasks
    ),
    ]

class linq.hooks.TaskStateChangeHook(*, parameters: HookParameters, filter: Literal[Filters.ON_TASK_STATE_CHANGE] = FieldInfo(annotation=NoneType, required=True, description='Event type triggering the hook. No longer required.', deprecated=True), task_ids: list[str] = FieldInfo(annotation=NoneType, required=True, description='List of task IDs that the hook should be triggered for. If empty, the hook will be triggered for all tasks.'))

Hook to be triggered when a task state changes.

3. SafetyStateChangeHook: Triggered when there is a change in the safety state of the workcell such as Safety alerts or Emergency stops.

hook = [
    SafetyStateChangeHook(
        parameters=HookParameters(
            url="http://example.com",
        )
    ),
]

class linq.hooks.SafetyStateChangeHook(*, parameters: HookParameters, filter: Literal[Filters.ON_SAFETY_STATE_CHANGE] = FieldInfo(annotation=NoneType, required=True, description='Event type triggering the hook. No longer required.', deprecated=True))

Hook to be triggered on changes to the state of the safety adapter.

4. LabwareMovementHook: Triggered when labware is moved within the workflow.

   Parameters:

• Labware_ids: List of labware IDs to monitor. Leave empty for all labware.

• Trigger_on: Specifies when to trigger the hook: ‘start’ (pickup), ‘end’ (dropoff), or ‘both’.

hook = [
    LabwareMovementHook(
        parameters=HookParameters(
            url="http://example.com",
        ),
        labware_ids=[],  # List of labware IDs, can be left empty
        trigger_on="both",  # Trigger on labware pickup, dropoff, or both
    ),
]

class linq.hooks.LabwareMovementHook(*, parameters: HookParameters, filter: Literal[Filters.ON_LABWARE_MOVEMENT] = FieldInfo(annotation=NoneType, required=True, description='Event type triggering the hook. No longer required.', deprecated=True), labware_ids: list[str] = FieldInfo(annotation=NoneType, required=True, description='List of labware IDs that the hook should be triggered for. If empty, the hook will be triggered for all labware.'), trigger_on: Literal['start', 'end', 'both'] = FieldInfo(annotation=NoneType, required=False, default='both', title='Trigger On', description="When to trigger the hook. 'start' triggers the hook when the labware is picked up, 'end' triggers the hook when the labware is placed down, 'both' does both."))

Hook to be triggered when labware is transported.

5. NewPlanHook: Triggered when a new plan is created by the scheduler.

hook = [
    NewPlanHook(
        parameters=HookParameters(
            url="http://example.com",
        )
    ),
]

class linq.hooks.NewPlanHook(*, parameters: HookParameters, filter: Literal[Filters.ON_NEW_PLAN] = FieldInfo(annotation=NoneType, required=True, description='Event type triggering the hook. No longer required.', deprecated=True))

Hook to be triggered when a new plan is generated.

Webhook requests

Hook Post requests have the following formats: