Slack

pip install "flyteplugins-slack[app]"

import flyte
from flyte.extras.webhooks import WebhookAppEnvironment, WebhookEvent, run_once
from flyteplugins.slack import SlackProvider, approval, events, notify

# SLACK_SIGNING_SECRET is mounted automatically. It's the signing secret from
# Basic Information, not the bot token, which goes on the task environment.
#
# `scopes` lists the channel IDs to act on: where the bot is mentioned, where
# /deploy is used, and where approvals are posted. Events from other channels
# are acknowledged but not dispatched.
app_env = WebhookAppEnvironment(
    name="slack-webhooks",
    providers=[SlackProvider()],
    scopes=["C0DEPLOYS"],
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-slack[app]"),
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

# Adds a handler that resolves the condition behind each button posted by
# `approval.request`. Each button's value carries the run, action, and
# condition names, so no other configuration is needed.
approval.register(app_env)

@app_env.on_event(events.AppMention.ANY)
async def on_mention(event: WebhookEvent) -> dict:
    """Launch a run for each @-mention.

    The dedupe key identifies one message. For one run per thread, build a
    key from `thread_ts` and pass it to `run_once` instead.
    """
    import flyte.remote as remote

    task = remote.Task.get(name="slack-bot.answer", auto_version="latest")
    slack_event = event.payload["event"]
    result = await run_once.aio(
        task,
        key=event.dedupe_key(),
        channel=event.scope,
        text=event.title or "",
        thread_ts=slack_event.get("thread_ts") or slack_event["ts"],
    )
    return {"run": result.run.name, "created": result.created}

@app_env.on_event(events.Command, action="/deploy")
async def on_deploy_command(event: WebhookEvent) -> dict:
    """Acknowledge the /deploy slash command.

    `respond` posts to the command's `response_url` and needs no bot token.
    Slack expects a reply within three seconds, so acknowledge here and do
    longer work in a launched run.
    """
    await notify.respond(event.payload["response_url"], "Deploy queued.")
    return {"ok": True}

# Every Block Kit button in the workspace
@app_env.on_event(events.Interaction.BLOCK_ACTIONS)
async def any_button(event): ...

# One button
@app_env.on_event(events.Interaction.BLOCK_ACTIONS, action="redeploy")
async def redeploy_button(event): ...

env = flyte.TaskEnvironment(
    name="slack-bot",
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-slack"),
    # The bot token (xoxb-...) from OAuth & Permissions. Posting requires the
    # `chat:write` scope, and the bot must be invited to the channel.
    secrets=[flyte.Secret(key="SLACK_BOT_TOKEN", as_env_var="SLACK_BOT_TOKEN")],
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

@env.task
async def answer(channel: str, text: str, thread_ts: str) -> str:
    """Post a threaded reply, then edit it when the work finishes.

    `post` returns the message's `ts`, which `update` uses to edit it.
    """
    ts = await notify.post(channel, f"Working on: {text}", thread_ts=thread_ts)
    await notify.update(channel, ts, f"Done: {text}")
    return ts

@env.task
async def deploy_with_approval(release: str, channel: str = "C0DEPLOYS") -> str:
    """Post approval buttons and wait for a click.

    `approval.request` posts Block Kit buttons and waits on a
    `flyte.new_condition`. The handler added by `approval.register` resolves
    the condition when someone clicks. The condition can also be resolved
    from the Flyte UI.
    """
    decision = await approval.request.aio(
        channel,
        f"Deploy `{release}` to prod?",
        options=["approve", "reject"],
        timeout=3600,
    )
    if decision != "approve":
        return f"{release}: not deployed ({decision})"
    return f"{release}: deployed"

# A separate environment with no secrets, so the replay runs before any
# secret is created.
replay_env = flyte.TaskEnvironment(
    name="slack-replay",
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-slack"),
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

@replay_env.task
async def replay_sample_delivery() -> dict[str, str]:
    """Verify and parse the sample delivery bundled with the plugin.

    The provider rejects deliveries more than five minutes old, so
    `SAMPLE_DELIVERY` signs the payload with the current time when called.
    """
    import flyteplugins.slack as plugin

    secret = "a-test-signing-secret"
    sign, body = plugin.SAMPLE_DELIVERY
    headers = sign(body, secret)

    assert plugin.verify(body, headers, secret), "a correctly signed delivery must verify"
    assert not plugin.verify(body, headers, "wrong-secret"), "a bad signature must not"

    event = plugin.parse(headers, body)
    return {
        "qualified_type": event.qualified_type,
        "scope": event.scope or "",
        "title": event.title or "",
        "actor": event.actor or "",
        "dedupe_key": event.dedupe_key(),
        # Maximum delivery age, in seconds.
        "max_request_age": str(plugin.MAX_REQUEST_AGE_SECONDS),
    }

flyte run --local slack_tasks.py replay_sample_delivery