Webhooks

A user can define webhooks which will send POST requests to the specified URL when certain conditions are met.

Defining Webhooks

Webhooks are defined at the workflow level. All hooks take a HookParameters field which includes the URL and header that should be used in the POST request.

pydantic model linq.hooks.HookParameters

Show JSON schema

{
   "title": "HookParameters",
   "type": "object",
   "properties": {
      "url": {
         "title": "Url",
         "type": "string"
      },
      "headers": {
         "additionalProperties": {
            "type": "string"
         },
         "description": "HTTP headers to include in the request",
         "title": "Headers",
         "type": "object"
      }
   },
   "required": [
      "url"
   ]
}

Fields:

headers (dict[str, str])
url (str)

field url: str [Required]

field headers: dict[str, str] [Optional]: HTTP headers to include in the request

to_api_input() → HookParametersInput

classmethod from_api_output(api_output_parameters: HookParameters) → Self

Currently, users may define the following hooks:

RunStateChangeHook
TaskStateChangeHook
SafetyStateChangeHook
LabwareMovementHook
NewPlanHook

RunStateChangeHook will send a POST request when the run changes state.

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 requires a task ID. A POST request will be sent when the specified task enters a new state.

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 will send a POST request when the safety state changes.

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 requires labware IDs (for specific labware items, otherwise leave blank for all labware) and a trigger field to be set (start, end or both). A POST request will be sent when the selected trigger event occurs.

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 will send a POST request when a new plan is created.

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:

field state: RunState [Required]

field message: str = ''

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:

field task_id: str [Required]

field instrument_id: str [Required]

field state: TaskState [Required]

field action: str [Required]

field error: str = ''

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:

state (str)

field state: str [Required]

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:

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]

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)

field remaining_worktime: int [Required]