Sending PyLabRobot notifications to Slack#

A basic example of sending Slack messages from Python, wired into a PLR protocol and PLR’s EventBus. The webhook helper uses only the Python standard library — for the full webhook API (formatting, attachments, threading, Block Kit), see Slack’s Incoming Webhooks docs.

Prerequisites#

1. Store the webhook URL#

Save the URL to ~/.plr_slack_webhook — outside the repo, mode 600 so only your user can read it:

umask 077 && echo "https://hooks.slack.com/services/..." > ~/.plr_slack_webhook

The helper below reads from this file. Pass webhook_url= to override (e.g. for testing).

2. A tiny stdlib notifier#

Slack’s incoming-webhook endpoint accepts a JSON POST. The standard library is enough — no httpx, requests, or slack_sdk needed.

import json
import urllib.request
import urllib.error
from pathlib import Path

WEBHOOK_FILE = Path("~/.plr_slack_webhook").expanduser()


def _load_webhook() -> str | None:
  if WEBHOOK_FILE.exists():
    return WEBHOOK_FILE.read_text().strip() or None
  return None


def slack_notify(text: str, webhook_url: str | None = None, timeout: float = 5.0) -> None:
  url = webhook_url or _load_webhook()
  if not url:
    print(f"[slack_notify: no webhook configured at {WEBHOOK_FILE}] {text}")
    return

  req = urllib.request.Request(
    url,
    data=json.dumps({"text": text}).encode("utf-8"),
    headers={"Content-Type": "application/json"},
    method="POST",
  )
  try:
    with urllib.request.urlopen(req, timeout=timeout) as resp:
      resp.read()
  except (urllib.error.URLError, TimeoutError) as e:
    print(f"[slack_notify failed: {e}] {text}")


slack_notify(":wave: Hello from PyLabRobot")

3. Wrap a PLR run#

Send a message at start, on success, and on failure. The finally block guarantees lh.stop() runs even if the protocol crashes.

import traceback

from pylabrobot.liquid_handling import LiquidHandler
from pylabrobot.liquid_handling.backends.hamilton.STAR_chatterbox import STARChatterboxBackend
from pylabrobot.resources.hamilton import STARLetDeck

# Using the chatterbox backend so this notebook runs without hardware.
# On a real machine, swap in `STARBackend()` (or whatever your machine uses).
lh = LiquidHandler(backend=STARChatterboxBackend(), deck=STARLetDeck())

try:
  await lh.setup()
  slack_notify(":test_tube: Protocol started")

  # ... your protocol goes here ...
  # await lh.pick_up_tips(...)
  # await lh.aspirate(...)
  # await lh.dispense(...)
  # await lh.drop_tips()

  slack_notify(":white_check_mark: Protocol finished")
except Exception as e:
  slack_notify(f":rotating_light: Protocol failed: `{e}`\n```{traceback.format_exc()}```")
  raise
finally:
  await lh.stop()

4. Forward semantic EventBus events#

PLR’s EventBus can automatically report instrument operations. This subscriber forwards completed and failed semantic operations while ignoring lower-level transport and firmware diagnostics.

EventBus subscribers run synchronously with PLR, so the callback only submits work to a background thread. A slow Slack request therefore does not delay the instrument operation.

from concurrent.futures import ThreadPoolExecutor

from pylabrobot.events import EventBus, PLREvent, use_event_bus


def semantic_slack_message(event: PLREvent) -> str | None:
  """Return Slack text for events we want to forward, or None to filter an event out."""
  # Semantic lifecycle events identify their high-level operation in context.
  # Lower-level transport, firmware, and other diagnostic events may not.
  operation = event.context.get("operation")
  if not isinstance(operation, str):
    return None

  # Forward outcomes only. Ignoring `.started` keeps Slack volume manageable while
  # still reporting every successful operation and every operation failure.
  status = event.name.removeprefix(f"{operation}.")
  if status not in {"completed", "failed"}:
    return None

  resources = event.data.get("resources", [])
  resource_names = [resource["name"] for resource in resources if "name" in resource]
  resource_text = f" ({', '.join(resource_names)})" if resource_names else ""

  if status == "failed":
    error = event.data.get("error_message", "unknown error")
    return f":rotating_light: `{operation}` failed{resource_text}: `{error}`"
  return f":white_check_mark: `{operation}` completed{resource_text}"


slack_worker = ThreadPoolExecutor(max_workers=1)
event_bus = EventBus()


def forward_semantic_event_to_slack(event: PLREvent) -> None:
  # EventBus subscribers run synchronously, so only queue network work here.
  message = semantic_slack_message(event)
  if message is not None:
    slack_worker.submit(slack_notify, message)


unsubscribe = event_bus.subscribe(forward_semantic_event_to_slack)
try:
  with use_event_bus(event_bus):
    await lh.setup()
    # Instrumented PLR operations in this scope are forwarded automatically.
    await lh.stop()
finally:
  unsubscribe()
  slack_worker.shutdown(wait=True)

5. Richer messages with Block Kit#

text is just one of the payload fields. Slack also accepts Block Kit for headers, sections, fields, and dividers:

def slack_post(payload: dict, webhook_url: str | None = None, timeout: float = 5.0) -> None:
  url = webhook_url or _load_webhook()
  if not url:
    print(f"[slack_post: no webhook configured at {WEBHOOK_FILE}] {payload}")
    return
  req = urllib.request.Request(
    url,
    data=json.dumps(payload).encode("utf-8"),
    headers={"Content-Type": "application/json"},
    method="POST",
  )
  try:
    with urllib.request.urlopen(req, timeout=timeout) as resp:
      resp.read()
  except (urllib.error.URLError, TimeoutError) as e:
    print(f"[slack_post failed: {e}] {payload}")


slack_post({
  "blocks": [
    {"type": "header", "text": {"type": "plain_text", "text": "Run complete"}},
    {"type": "section", "fields": [
      {"type": "mrkdwn", "text": "*Plates:*\n4"},
      {"type": "mrkdwn", "text": "*Duration:*\n00:42:13"},
    ]},
  ]
})

Notes#

  • The webhook URL lives in ~/.plr_slack_webhook, outside the repo. Anyone with the URL can post to that channel, so keep it mode 600.

  • The helper catches URLError and TimeoutError so a failed POST doesn’t abort the protocol.

  • The EventBus example forwards semantic operation outcomes, not .started or diagnostic events. Adapt that filter to the notification volume you need.

  • Keep EventBus subscribers fast. For production use, enqueue network or disk work rather than performing it directly in the callback.

  • Incoming Webhooks rate-limit to ~1 msg/sec per channel.

  • Webhooks are write-only. Slack → robot (e.g. a /pause command) needs the Events API.