Slack notifications and operator confirmation

The workflow posts a Slack message, then pauses until an operator confirms in UniteLabs.

You add one workflow package to a clone of the workflow template and deploy it. On this main path the operator confirms in UniteLabs, and replies, reactions, and buttons in Slack do nothing. Answer from Slack instead of the platform shows a short variant where the operator answers with an emoji reaction in Slack.

Why the workflow has to be deployed. A script started locally uses a temporary engine that ends with the process, so the platform never sees a run to confirm. HITL basics requires a workflow engine connected to the platform.

Two ways to get the workflow onto the platform. You can either create the workflow directly in the platform UI or develop it locally in your IDE and deploy it using your credentials. The platform UI requires no repository clone, package-registry access, local credentials, or deploy command, but it does not provide version control. The local approach keeps the workflow code in a cloned repository and deploys it using the template script. Steps 2, 3, and 6 apply to both. Steps 1, 4, and 5 are the template path; Alternative: write the workflow in the platform replaces them.

Before you start

You need:

• Python 3.12 or newer and uv.
• Access to the UniteLabs package registry, set up as described in Set up registry access.
• A UniteLabs account that can create secrets, deploy workflows, and confirm runs, plus API client credentials (BASE_URL, AUTH_URL, CLIENT_ID, CLIENT_SECRET). Registry credentials and platform credentials are separate.
• A Slack workspace where you can install an app, and a channel for test messages. For a private channel you must already be a member. If your workspace requires administrator approval for apps, arrange that first.

Use Get access if you are missing UniteLabs credentials.

Work through Your first workflow first. It leaves you with the clone and the environment this guide builds on.

Step 1: Get the workflow template

Clone the template:

git clone https://gitlab.com/unitelabs/workflows/workflow-template.git
cd workflow-template

Every top-level w*-*/ directory is a standalone workflow package. The clone ships four, so the next free number is w05. Check the numbering in your clone before choosing a name.

Copy .env.example to .env in the repo root and fill in the four values for your tenant:

BASE_URL=<your-tenant-api-base-url>
AUTH_URL=<your-tenant-authentication-url>
CLIENT_ID=<your-client-id>
CLIENT_SECRET=<your-client-secret>

Use the API and authentication URLs supplied for your tenant, not the browser address. BASE_URL has no /v1 suffix; the client appends the API version. URL examples are in Authenticate Client.

The deploy script reads this file. Keep it private; the template ignores it in Git.

Expected result: A clone of the template, and a .env in its root holding the four connection settings.

Step 2: Create a Slack incoming webhook

An incoming webhook is a URL that accepts messages for one Slack channel.

1. Open Your Apps, select Create New App, then From scratch.
2. Enter a name such as UniteLabs Notifier, select your workspace, and create the app.
3. Open Incoming Webhooks in the app settings and enable Activate Incoming Webhooks.
4. Select Add New Webhook to Workspace, choose your test channel, and authorize the installation.
5. Back in the app settings, find Webhook URLs for Your Workspace. Check the channel and copy the new webhook URL for the next step.

Expected result: Slack lists a webhook for the intended channel.

This example needs no bot token. The URL is a credential: anyone who has it can post to that channel. Store it in the platform secret, never in workflow code. See Slack's incoming webhook guide.

Step 3: Save the webhook as a platform secret

Open UniteLabs in your browser and select the same tenant your .env points to.

1. Open Secrets using the padlock entry in the navigation, then select Create.
2. Name the secret exactly slack-webhook. If that name already exists, check what uses it before reusing it; do not overwrite another integration's credential.
3. Select the type Slack Webhook, paste the webhook URL into the Webhook URL field, and save.
4. Return to the Secrets list and check that slack-webhook appears.

Expected result: The tenant holds a secret named slack-webhook whose value is the webhook URL.

A saved secret cannot be edited. To change a value, delete the secret and create it again under the same name.

Slack Webhook stores the URL in url. Do not use the generic Webhook type; it describes an arbitrary HTTP call and requires a method and headers.

The code in Step 4 loads this type by name through SlackWebhook. Other types use different fields and cannot be loaded that way.

Notify Type and Allow Private Urls are optional. Notify Type sets the Slack message color (info, success, warning, or failure; default info). Allow Private Urls permits private-network URLs and is enabled by default. Leave both unchanged.

Secrets explains the underlying API and the type schemas. Creating the secret in the interface needs no API request.

Step 4: Add the notification workflow

Create the directory w05-slack-notify/ in the repo root, next to the workflows the template ships. It holds three files:

w05-slack-notify/
├── pyproject.toml
└── src/
    └── w05_slack_notify/
        ├── __init__.py
        └── workflow.py

Place the Python files in src/w05_slack_notify/, not directly in src/. The deploy script derives the directory name from the slug by replacing hyphens with underscores; the build backend expects the same layout.

Create __init__.py to mark the directory as a Python package:

"""Workflow 05: Slack Notifier Demo. See workflow.py for the @workflow entrypoint."""

from importlib.metadata import version

__version__ = version("w05-slack-notify")

Write pyproject.toml:

[project]
name = "w05-slack-notify"
version = "0.1.0"
description = "Posts a Slack message, then waits for operator confirmation."
requires-python = ">=3.12"
dependencies = [
    "unitelabs-sdk[automate]",
    "httpx",
]

[tool.unitelabs.workflow]
display_name = "Slack confirmation demo"
entrypoint = "workflow.py:slack_notify_flow"
tags = ["slack", "hitl"]

[tool.uv]
prerelease = "if-necessary-or-explicit"

[[tool.uv.index]]
name = "unitelabs"
url = "https://gitlab.com/api/v4/groups/1009252/-/packages/pypi/simple"
authenticate = "always"

[tool.uv.sources]
unitelabs-sdk = { index = "unitelabs" }

[build-system]
requires = ["uv_build>=0.4,<0.10"]
build-backend = "uv_build"

The deploy script reads [tool.unitelabs.workflow]: display_name is the platform name and entrypoint names the file and function to start. [project].name is the deploy slug.

The SDK requirement carries no version, so you get the current release. The code below needs operator_confirm, which the SDK exports from 0.13.0 onwards; 0.12 does not have it. Keep that in mind if you reuse a workflow package that pins an older version.

Now write src/w05_slack_notify/workflow.py:

import httpx
from prefect.blocks.notifications import SlackWebhook
from unitelabs.sdk import get_logger, operator_confirm, phase, step, workflow

SECRET_NAME = "slack-webhook"


@step(name="Notify Slack")
async def notify_slack(text: str) -> None:
    webhook = await SlackWebhook.aload(SECRET_NAME)
    async with httpx.AsyncClient() as http:
        response = await http.post(webhook.url.get_secret_value(), json={"text": text})
        response.raise_for_status()
    get_logger().info("Slack notification sent.")


@phase(name="Request confirmation")
async def request_confirmation() -> None:
    await notify_slack(
        "Slack confirmation demo: open UniteLabs, find the waiting run, "
        "and confirm the test request."
    )
    await operator_confirm("Confirm that you received the Slack test message.", timeout=600)


@workflow(name="Slack confirmation demo")
async def slack_notify_flow() -> None:
    await request_confirmation()
    get_logger().info("Confirmation received; workflow complete.")

notify_slack() loads the secret and posts the message without exposing the URL in code.

await webhook.anotify(text) uses the engine notification layer, which adds a fixed Prefect Notifications footer and a colored attachment. This example posts to the webhook directly and can therefore use any Slack message payload. See the Workflow, Phase, Step taxonomy when splitting larger workflows.

operator_confirm() follows HITL basics. This example allows ten minutes (timeout=600) and performs no physical action. The SDK sets step retries to 0, so a failed send is not retried automatically.

This package omits __main__.py: the secret and confirmation require a platform run. The platform starts the function named in entrypoint.

Install the new package:

uv sync --directory w05-slack-notify

Expected result: The command creates w05-slack-notify/.venv and a lockfile without reporting a dependency conflict.

Step 5: Deploy the workflow

From the repo root:

uv run scripts/deploy.py w05-slack-notify

The argument is [project].name. The script bundles the workflow with the template's shared/ library and creates or updates the platform workflow. Deploy a workflow documents other forms, including --channel dev.

Expected result: The script prints the display name, the resolved entrypoint, the bundle size, and whether it created or updated the record. The workflow then appears under Workflows in the platform.

If the upload rejects the SDK's @workflow entrypoint, replace both SDK decorators with flow and deploy again:

from prefect import flow

@flow(name="Request confirmation")
async def request_confirmation() -> None:
    ...

@flow(name="Slack confirmation demo")
async def slack_notify_flow() -> None:
    ...

Swap the pair together. @phase reads a context file that only @workflow writes, so a @flow entrypoint combined with a @phase crashes on the first phase call. The template's authoring rules state the same constraint. Keep the SDK decorators if the deploy succeeds.

Step 6: Run the workflow and confirm the request

1. Open Workflows in the platform and select Slack confirmation demo.
2. Select Run. The run appears under Runs.

Expected result in Slack: A message from your app asking you to confirm the waiting run.

Expected result in the platform: The run reaches AWAITING_INPUT. Open it and submit the confirmation. The input has a single field, confirmed, already set to true.

Expected result after the confirmation: The run reaches COMPLETED, and its log shows Slack notification sent. followed by Confirmation received; workflow complete. Without a confirmation within ten minutes, the run fails on the timeout.

Run a workflow covers starting runs from Python and from the REST API, which is the way to trigger this workflow from an existing system.

Alternative: write the workflow in the platform

This path replaces Steps 1, 4, and 5. You still need the webhook and secret.

1. Open Workflows and select Create.
2. Enter the name Slack confirmation demo. Tags and description are optional. Leave Runtime Image empty to use the platform default.
3. Under Dependencies, add two entries: unitelabs-sdk[automate] and httpx. The field takes the same syntax as pip install, so extras and version specifiers both work; without a version you get the current release. The platform installs these packages for the run, which is why this path needs no registry access from you.
4. Leave the Files upload empty and select Create. The platform creates the workflow with a file workflow.py holding a placeholder flow.
5. Open the new workflow, select the Code tab, and replace the content of workflow.py with the code below. Save it.

import httpx
from prefect.blocks.notifications import SlackWebhook
from unitelabs.sdk import get_logger, operator_confirm, phase, step, workflow

SECRET_NAME = "slack-webhook"


@step(name="Notify Slack")
async def notify_slack(text: str) -> None:
    webhook = await SlackWebhook.aload(SECRET_NAME)
    async with httpx.AsyncClient() as http:
        response = await http.post(webhook.url.get_secret_value(), json={"text": text})
        response.raise_for_status()
    get_logger().info("Slack notification sent.")


@phase(name="Request confirmation")
async def request_confirmation() -> None:
    await notify_slack(
        "Slack confirmation demo: open UniteLabs, find the waiting run, "
        "and confirm the test request."
    )
    await operator_confirm("Confirm that you received the Slack test message.", timeout=600)


@workflow(name="Slack confirmation demo")
async def entrypoint() -> None:
    await request_confirmation()
    get_logger().info("Confirmation received; workflow complete.")

This is the Step 4 code with the last function renamed to entrypoint, which matches the record created by the form.

Expected result: The workflow appears under Workflows, its Code tab shows your workflow.py, and its dependency list shows the SDK entry. Continue with Step 6.

Files in the Code tab share one directory and can import each other directly. This path has no package layout or __init__.py; uploading a ZIP instead uses the template layout, including pyproject.toml.

Add the notification to an existing workflow

Copy notify_slack() and its imports into your workflow, then call it before operator_confirm(). Name the action, run, and confirmation location in the message. Keep the webhook in the platform secret.

When several workflows need the notification, move the function into the template's shared/ library, under shared/src/shared/steps/, and import it from there. The deploy script ships shared/ with every bundle, so the import resolves on the platform exactly as it does locally. A workflow that imports from shared has to declare it, which the pyproject.toml above does not: add "shared" to [project].dependencies and shared = { path = "../shared", editable = true } under [tool.uv.sources], then run uv sync --directory shared once before you sync the workflow. The workflows the template ships show both lines.

Troubleshooting

Answer from Slack instead of the platform

An incoming webhook only delivers a message. Somebody has to read the answer, and Slack pushes button clicks only to a server that listens on a public URL or holds a Socket Mode connection. Emoji reactions need no such server: they are readable through the Web API, so the workflow posts the question and then checks its own message until an agreed reaction appears.

This replaces the notification step and incoming webhook; reading reactions needs the message timestamp, which a webhook does not return.

1. Open the Slack app from Step 2 at Your Apps and give it two more permissions, chat:write to post and reactions:read to see the reactions. There are two places to do this, and both end up in the same app configuration:
   • OAuth & Permissions in the sidebar, under Features. Scroll past the token boxes and the redirect URLs to the Scopes section, then add the two entries under Bot Token Scopes.
   • App Manifest in the sidebar. Add the two entries to the existing list under oauth_config.scopes.bot and save. Do not replace the rest of the manifest, or the incoming webhook from Step 2 disappears with it.

   oauth_config:
     scopes:
       bot:
         - incoming-webhook
         - chat:write
         - reactions:read
   

2. Changing the permissions invalidates the installation, so install the app again. Slack shows a banner asking for it; the button sits at the top of OAuth & Permissions and reads Reinstall to Workspace. After that, the same page shows the Bot User OAuth Token, which starts with xoxb-. Copy it.
3. In your Slack channel, invite the bot with /invite @UniteLabs Notifier. It has to post there, and posting requires membership. This has to be a channel, not a direct message.
4. In the platform, create a second secret named slack-bot-token of type Slack Credentials, and paste the token into its token field.
5. Add the package prefect-slack to the workflow, next to httpx. It holds the SlackCredentials class that reads the secret. On the platform path that means one more entry in the Dependencies field of the workflow; on the template path one more line in [project].dependencies, followed by uv sync. Without it the run fails while loading the code, with ModuleNotFoundError: No module named 'prefect_slack'.
6. Replace the notification step with the one below, and set CHANNEL to your channel. Keep the name of the last function as your path requires: entrypoint on the platform path, or whatever [tool.unitelabs.workflow].entrypoint names on the template path. The example below uses entrypoint.

import asyncio

import httpx
from prefect_slack import SlackCredentials
from unitelabs.sdk import get_logger, phase, step, workflow

TOKEN_SECRET = "slack-bot-token"
CHANNEL = "#lab-notifications"
APPROVE = {"white_check_mark", "heavy_check_mark"}
REJECT = {"x", "no_entry"}
POLL_SECONDS = 10
ATTEMPTS = 60


@step(name="Ask in Slack and wait")
async def ask_in_slack(question: str) -> str:
    credentials = await SlackCredentials.aload(TOKEN_SECRET)
    token = credentials.token.get_secret_value()
    logger = get_logger()

    async with httpx.AsyncClient(
        base_url="https://slack.com/api",
        headers={"Authorization": f"Bearer {token}"},
    ) as slack:
        posted = await slack.post(
            "/chat.postMessage",
            json={
                "channel": CHANNEL,
                "text": f"{question}\n\nReact with :white_check_mark: to continue or :x: to abort.",
            },
        )
        posted.raise_for_status()
        result = posted.json()
        if not result.get("ok"):
            raise ValueError(f"Slack rejected the message: {result.get('error')}")
        channel_id, message_ts = result["channel"], result["ts"]
        logger.info("Question posted to Slack.")

        for _ in range(ATTEMPTS):
            await asyncio.sleep(POLL_SECONDS)
            read = await slack.get(
                "/reactions.get",
                params={"channel": channel_id, "timestamp": message_ts},
            )
            read.raise_for_status()
            body = read.json()
            if not body.get("ok"):
                raise ValueError(f"Slack rejected the read: {body.get('error')}")
            for reaction in body.get("message", {}).get("reactions", []):
                name = reaction["name"].split("::")[0]
                if name in REJECT:
                    raise RuntimeError(f"Aborted in Slack with :{name}:.")
                if name in APPROVE:
                    logger.info(f"Confirmed in Slack with :{name}: by {reaction['users']}.")
                    return name

    raise TimeoutError(f"No reaction within {POLL_SECONDS * ATTEMPTS} seconds.")


@phase(name="Request confirmation")
async def request_confirmation() -> None:
    await ask_in_slack("Slack reaction demo: a workflow is waiting for your go-ahead.")


@workflow(name="Slack reaction demo")
async def entrypoint() -> None:
    await request_confirmation()
    get_logger().info("Confirmation received; workflow complete.")

chat.postMessage returns the channel ID and the message timestamp, and the two together address the message for reactions.get.

Emoji names carry no colons. ✅ is white_check_mark, and heavy_check_mark is the narrower ✔️, which is why APPROVE holds both. For emoji with skin tones Slack appends ::skin-tone-3, which the split removes. A reaction from REJECT fails the step, so the run ends as failed.

Ten minutes of checking every ten seconds is the timeout; a run that nobody answers fails with the TimeoutError.

What this version gives up:

• The run never pauses. The platform shows it as running instead of waiting for input, so there is no input form and the answer is not recorded with the run. Resource use is not the difference: in the SDK version checked here, operator_confirm() registers the paused state on the platform and then waits in place, asking every few seconds whether the run has been resumed. Both variants therefore occupy their worker while they wait.
• Anyone in the channel can react. The code logs the reaction's users but does not check whether they are allowed to answer, and Slack states that this list may be incomplete while count is not.
• The answer arrives with the next check, up to ten seconds late.
• Approval and rejection at the same time are not handled. Whichever reaction the loop sees first wins.

For buttons, forms, and typed answers, the person's action has to reach a listening service, which then resumes the paused run through the API. The platform side of that is described in HITL basics: Pattern 3, Resume via API: send confirmed: true under the waiting phase ID from the run's current status. Other required fields have to match the schema described in Typed operator inputs. That service needs its own credentials and its own code for authorization, duplicate answers, stale requests, and API errors, and it has to run somewhere. It is a separate piece of software from the workflow.