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:

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.

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.

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.

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.

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:

pydantic model linq.maestro_webhooks_temp.RunStateChangeHookRequest

Request to be sent by the hook when a run state changes.

Show JSON schema

{
   "title": "RunStateChangeHookRequest",
   "description": "Request to be sent by the hook when a run state changes.",
   "type": "object",
   "properties": {
      "run_id": {
         "title": "Run Id",
         "type": "string"
      },
      "timestamp": {
         "description": "Timestamp of the event in RFC3339 format",
         "title": "Timestamp",
         "type": "string"
      },
      "state": {
         "$ref": "#/$defs/RunState"
      },
      "message": {
         "default": "",
         "title": "Message",
         "type": "string"
      }
   },
   "$defs": {
      "RunState": {
         "enum": [
            "started",
            "paused",
            "resumed",
            "stopped"
         ],
         "title": "RunState",
         "type": "string"
      }
   },
   "required": [
      "run_id",
      "timestamp",
      "state"
   ]
}

Config:

use_enum_values: bool = True

Fields:

message (str)
run_id (str)
state (linq.maestro_webhooks_temp.RunState)
timestamp (str)

field state: RunState [Required]

field message: str = ''

field run_id: str [Required]

field timestamp: str [Required]: Timestamp of the event in RFC3339 format

pydantic model linq.maestro_webhooks_temp.TaskStateChangeHookRequest

Request to be sent by the hook when a task state changes.

Show JSON schema

{
   "title": "TaskStateChangeHookRequest",
   "description": "Request to be sent by the hook when a task state changes.",
   "type": "object",
   "properties": {
      "run_id": {
         "title": "Run Id",
         "type": "string"
      },
      "timestamp": {
         "description": "Timestamp of the event in RFC3339 format",
         "title": "Timestamp",
         "type": "string"
      },
      "task_id": {
         "title": "Task Id",
         "type": "string"
      },
      "instrument_id": {
         "title": "Instrument Id",
         "type": "string"
      },
      "state": {
         "$ref": "#/$defs/TaskState"
      },
      "action": {
         "title": "Action",
         "type": "string"
      },
      "error": {
         "default": "",
         "title": "Error",
         "type": "string"
      }
   },
   "$defs": {
      "TaskState": {
         "enum": [
            "started",
            "failed",
            "succeeded"
         ],
         "title": "TaskState",
         "type": "string"
      }
   },
   "required": [
      "run_id",
      "timestamp",
      "task_id",
      "instrument_id",
      "state",
      "action"
   ]
}

Config:

use_enum_values: bool = True

Fields:

action (str)
error (str)
instrument_id (str)
run_id (str)
state (linq.maestro_webhooks_temp.TaskState)
task_id (str)
timestamp (str)

field task_id: str [Required]

field instrument_id: str [Required]

field state: TaskState [Required]

field action: str [Required]

field error: str = ''

field run_id: str [Required]

field timestamp: str [Required]: Timestamp of the event in RFC3339 format

pydantic model linq.maestro_webhooks_temp.SafetyStateChangeHookRequest

Request to be sent by the hook on changes to the state of the safety adapter.

Show JSON schema

{
   "title": "SafetyStateChangeHookRequest",
   "description": "Request to be sent by the hook on changes to the state of the safety adapter.",
   "type": "object",
   "properties": {
      "run_id": {
         "title": "Run Id",
         "type": "string"
      },
      "timestamp": {
         "description": "Timestamp of the event in RFC3339 format",
         "title": "Timestamp",
         "type": "string"
      },
      "state": {
         "title": "State",
         "type": "string"
      }
   },
   "required": [
      "run_id",
      "timestamp",
      "state"
   ]
}

Fields:

run_id (str)
state (str)
timestamp (str)

field state: str [Required]

field run_id: str [Required]

field timestamp: str [Required]: Timestamp of the event in RFC3339 format

pydantic model linq.maestro_webhooks_temp.LabwareMovementHookRequest

Request to be sent by the hook when labware is transported.

Show JSON schema

{
   "title": "LabwareMovementHookRequest",
   "description": "Request to be sent by the hook when labware is transported.",
   "type": "object",
   "properties": {
      "run_id": {
         "title": "Run Id",
         "type": "string"
      },
      "timestamp": {
         "description": "Timestamp of the event in RFC3339 format",
         "title": "Timestamp",
         "type": "string"
      },
      "labware_id": {
         "title": "Labware Id",
         "type": "string"
      },
      "state": {
         "$ref": "#/$defs/LabwareMovementState"
      },
      "source_instrument_id": {
         "title": "Source Instrument Id",
         "type": "string"
      },
      "destination_instrument_id": {
         "title": "Destination Instrument Id",
         "type": "string"
      },
      "source_slot": {
         "title": "Source Slot",
         "type": "integer"
      },
      "destination_slot": {
         "title": "Destination Slot",
         "type": "integer"
      }
   },
   "$defs": {
      "LabwareMovementState": {
         "enum": [
            "started",
            "finished",
            "failed"
         ],
         "title": "LabwareMovementState",
         "type": "string"
      }
   },
   "required": [
      "run_id",
      "timestamp",
      "labware_id",
      "state",
      "source_instrument_id",
      "destination_instrument_id",
      "source_slot",
      "destination_slot"
   ]
}

Fields:

destination_instrument_id (str)
destination_slot (int)
labware_id (str)
run_id (str)
source_instrument_id (str)
source_slot (int)
state (linq.maestro_webhooks_temp.LabwareMovementState)
timestamp (str)

field labware_id: str [Required]

field state: LabwareMovementState [Required]

field source_instrument_id: str [Required]

field destination_instrument_id: str [Required]

field source_slot: int [Required]

field destination_slot: int [Required]

field run_id: str [Required]

field timestamp: str [Required]: Timestamp of the event in RFC3339 format

pydantic model linq.maestro_webhooks_temp.NewPlanHookRequest

Request to be sent by the hook when a new plan is generated.

Show JSON schema

{
   "title": "NewPlanHookRequest",
   "description": "Request to be sent by the hook when a new plan is generated.",
   "type": "object",
   "properties": {
      "run_id": {
         "title": "Run Id",
         "type": "string"
      },
      "timestamp": {
         "description": "Timestamp of the event in RFC3339 format",
         "title": "Timestamp",
         "type": "string"
      },
      "remaining_worktime": {
         "title": "Remaining Worktime",
         "type": "integer"
      }
   },
   "required": [
      "run_id",
      "timestamp",
      "remaining_worktime"
   ]
}

Fields:

remaining_worktime (int)
run_id (str)
timestamp (str)

field remaining_worktime: int [Required]

field run_id: str [Required]

field timestamp: str [Required]: Timestamp of the event in RFC3339 format