Tri
11/13/2025, 5:21 PMMarvin
11/13/2025, 5:21 PMMarvin
11/13/2025, 5:32 PMPREFECT_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)
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)
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)
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.