<@ULVA73B9P> in prefect 3, can I create work pool,...
# ask-marvin
t
@Marvin in prefect 3, can I create work pool, automation, block through python API, instead of doing so in command line and the UI?
m
thought for 661.2 seconds
Short answer: yes. In Prefect 3.x you can create work pools, automations, and blocks fully via the Python API (no CLI/UI required). Below are minimal examples and the relevant docs. Authentication (works for Cloud or a self-hosted API): - Set
PREFECT_API_URL
and
PREFECT_API_KEY
in your environment before running the code - Docs: Connect to Prefect Cloud Create a work pool (example: a “process” pool)
Copy code
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import WorkPoolCreate

with get_client(sync_client=True) as client:
    pool = client.create_work_pool(
        WorkPoolCreate(
            name="my-process-pool",
            type="process",          # e.g. "process", "docker", "kubernetes", etc.
            description="Created via Python API",
            # base_job_template can be omitted for defaults on many types
        )
    )
    print(pool.name, pool.id)
Create a block (example: Slack webhook block)
Copy code
from prefect.blocks.notifications import SlackWebhook

# Save once; then reference by name elsewhere
SlackWebhook(url="<https://hooks.slack.com/services/XXX/YYY/ZZZ>").save(
    name="my-slack",
    overwrite=True,
)
Create an automation (example: notify on failed flow runs)
Copy code
from prefect.automations import Automation, EventTrigger, SendNotification, Posture

# This automation reacts to any failed flow run and sends a Slack notification
automation = Automation(
    name="Notify on failed flow runs",
    trigger=EventTrigger(
        # Match on the event type emitted for failed flow runs
        match={"event": "prefect.flow-run.Failed"},
        posture=Posture.Reactive,
    ),
    actions=[
        # Reference the saved Slack block by name
        SendNotification(
            block_document_name="slack-webhook/my-slack",
            body="Flow run {{ event.resource.name }} failed. See {{ event.related.url }}",
        )
    ],
)

# Synchronous creation (use .acreate() if you're in async code)
automation.create()
Notes and caveats: - The client will pick up your API URL/key from environment or Prefect settings. - You can use the async client too:
async with get_client() as client: ...
- Work pool types may require a
base_job_template
depending on the worker type; defaults are available for many built-ins. - Automations support many triggers (event/metric/composite) and actions (run deployment, send notification, pause/resume, etc.). See docs for additional fields and templates. Useful references: - Python client (work pools, queues, etc.): Client API (orchestration) - Automations (concepts and code): Creating automations - Blocks (concepts and saving/loading): Blocks overview If you can share: - the worker type you want for the work pool (process, docker, k8s, …), and - the exact automation trigger/action you want (e.g., which events, what message, any filters), I can tailor the snippets precisely to your use case.