Great goal — tagging the ACI container groups makes cleanup much easier. Short answer: - Prefect’s Azure Container Instance worker (Prefect 3.x) does not expose the worker’s ID/name to the job template or to the running container. There’s no built-in way to tag ACIs with the creating worker’s ID out of the box. - You can, however, add Azure tags via the work pool’s base job template (by editing the ARM template) and map them from

labels

(which you control via

job_variables

). If you truly need the worker ID, the supported path is to create a small custom worker that injects the tag. Options you have 1) Add tags via the ARM template in your work pool - The ACI worker’s base job template includes an

arm_template

that defines the Container Group. You can add a

tags

block there. - Easiest pattern: wire Azure tags to Prefect’s

labels

so anything you set in

job_variables.labels

becomes Azure tags. Steps: - Pull the default template:

prefect work-pool get-default-base-job-template --type azure-container-instance > aci.json

- Edit aci.json. In

job_configuration.arm_template.resources[0]

(the container group resource), add:

"tags": {{ labels }}

- Update your pool:

prefect work-pool update <your-pool-name> --base-job-template aci.json

- Now set labels at deploy/run time; they will become Azure tags:

python
deployment = flow.from_source(...).deploy(
    name="my-deployment",
    work_pool_name="<your-pool-name>",
    job_variables={
        "labels": {
            "cleanup": "true",
            "owner": "data-eng",
            "purpose": "prefect-run"
        }
    }
)

Notes: -

labels

is available in the ACI job template by default; mapping it to Azure

tags

works well because both are string → string maps. - This approach won’t give you the worker’s ID (it isn’t available in the Jinja context), but gets you useful tags for cleanup. Docs: - Azure Container Instance worker - Customize job variables - Workers concept 2) Tag with flow-run metadata from inside the flow If tagging with the worker ID isn’t strictly required, a common pattern is to tag the container group with the flow run ID (or deployment) after the container is up. Inside your flow, you can read

prefect.runtime.flow_run.id

and update ACI tags using the Azure SDK:

from prefect import flow
from prefect.runtime import flow_run

# pip install azure-identity azure-mgmt-containerinstance
from azure.identity import DefaultAzureCredential
from azure.mgmt.containerinstance import ContainerInstanceManagementClient

@flow
def tag_container_group(subscription_id: str, resource_group: str, container_group_name: str):
    cred = DefaultAzureCredential()
    client = ContainerInstanceManagementClient(cred, subscription_id)

    # Get the container group
    cg = client.container_groups.get(resource_group, container_group_name)

    # Update tags
    tags = cg.tags or {}
    tags["prefect-flow-run-id"] = flow_run.id.get()

    # Overwrite with updated tags
    client.container_groups.begin_create_or_update(
        resource_group,
        container_group_name,
        {
            "location": cg.location,
            "tags": tags,
            "properties": cg.properties
        }
    ).result()

Tip: Make the

container_group_name

deterministic in your base job template (e.g., incorporate

{{ name }}

or another unique string) so the flow knows which ACI group to update. Docs: - prefect.runtime.flow_run 3) If you must tag with the worker’s ID/name

- This is not available in the standard ACI worker’s template or runtime. The supported way is to implement a small custom worker that extends the ACI worker and injects

tags

with the worker’s identity before creating the container group. - High-level: subclass the ACI worker, override the part that builds/submits the ARM template, and add

tags={"prefect-worker-id": self.name}

(or similar). Then register this as a new worker type and create a work pool of that type. Docs: - Developing a custom worker Questions to confirm - Is tagging with the flow run ID acceptable for your cleanup job, or do you specifically need the worker’s ID? If worker ID is required, I can outline a minimal custom worker skeleton you can drop into your project.

Short version: Prefect’s ACI worker will try to delete the container group at the end of the run, but if the worker dies mid-run you can end up with orphaned ACIs. The most reliable approach is to run a small, periodic “reaper” flow (on a separate, stable worker) that lists ACI container groups and deletes any that are finished. Recommended setup 1) Tag ACIs created by Prefect so you can safely target them - Edit your ACI work pool’s base job template to pass Prefect

labels

through as Azure tags: - Export the default template:

prefect work-pool get-default-base-job-template --type azure-container-instance > aci.json

- In aci.json, add

"tags": {{ labels }}

to the container group resource:

"resources": [
  {
    "type": "Microsoft.ContainerInstance/containerGroups",
    "apiVersion": "2022-09-01",
    "name": "[parameters('container_group_name')]",
    "location": "[parameters('location')]",
    "tags": {{ labels }},
    "properties": {
      ...

- Update your pool:

prefect work-pool update <your-pool-name> --base-job-template aci.json

- Set labels when you deploy so they become Azure tags:

python
deployment = flow.from_source(...).deploy(
    name="my-deployment",
    work_pool_name="<your-pool-name>",
    job_variables={
        "labels": {
            "prefect-aci": "true",
            "team": "data-eng"
        }
    }
)

Docs: - Azure Container Instance worker - Customize job variables 2) Run a scheduled cleanup flow on a resilient worker - Host this on a VM/Process/Docker worker (not the ACI worker) so it still runs if your ACI worker is down. - It filters ACI groups by tag/prefix, checks that all containers have finished, and deletes the group. Example cleanup flow: ``` from datetime import datetime, timezone, timedelta from prefect import flow, get_run_logger # pip install azure-identity azure-mgmt-containerinstance from azure.identity import DefaultAzureCredential from azure.mgmt.containerinstance import ContainerInstanceManagementClient def group_finished(cg): # Returns True if no containers are running containers = getattr(cg, "containers", []) or [] for c in containers: iv = getattr(c, "instance_view", None) state = getattr(getattr(iv, "current_state", None), "state", None) if state in ("Running", "Pending", "Starting"): return False return True def group_old_enough(cg, min_age_minutes): # Uses the latest finish_time or start_time across containers containers = getattr(cg, "containers", []) or [] latest_time = None for c in containers: iv = getattr(c, "instance_view", None) cs = getattr(iv, "current_state", None) finish = getattr(cs, "finish_time", None) start = getattr(cs, "start_time", None) t = finish or start if t and (latest_time is None or t > latest_time): latest_time = t if latest_time is None: # no timing info; treat as old enough return True return datetime.now(timezone.utc) - latest_time >= timedelta(minutes=min_age_minutes) @flow def cleanup_aci( subscription_id: str, resource_group: str, tag_key: str = "prefect-aci", tag_value: str = "true", min_age_minutes: int = 10, ): log = get_run_logger() cred = DefaultAzureCredential() client = ContainerInstanceManagementClient(cred, subscription_id) to_delete = [] for cg in client.container_groups.list_by_resource_group(resource_group): tags = getattr(cg, "tags", {}) or {} if tags.get(tag_key) != tag_value: continue if group_finished(cg) and group_old_enough(cg, min_age_minutes): to_delete.append(cg.name) for name in to_delete: log.info(f"Deleting ACI container group: {name}")

client.container_groups.begin_delete(resource_group, name).result() log.info(f"Deleted {len(to_delete)} container group(s).") ``` Schedule this flow (e.g., every 10–30 minutes). It will clean up any completed ACIs, even if the ACI worker is offline. 3) Optional: trigger-based cleanup - If you’re using Prefect Cloud, you can also trigger the cleanup flow via an Automation when your ACI worker goes offline or when flows complete, but a periodic schedule is usually sufficient and simpler. Advanced (optional) - Sidecar “reaper” container: You can add a second container into the ACI group that watches the main container and deletes the group on completion. This works, but requires customizing the ARM template, container image, and identity permissions. The periodic cleanup flow is simpler and more maintainable. If you want, I can help you: - Patch your current work pool template to add tags, and - Package the cleanup flow into a small deployment YAML you can deploy right away.

Short answer: it’s generally safe. If your reaper deletes a completed ACI container group before the ACI worker does, the worst you’ll see is a benign 404/NotFound in the worker logs. It won’t flip a completed flow run back to a failure. The flow’s final state is written by the engine inside the container before exit; deleting the container group after that just short-circuits the worker’s own cleanup. To make it rock-solid: - Add a small grace period before deletion (e.g., 5–10 minutes) and only delete groups whose containers are terminated/succeeded. - Handle idempotency by catching 404s when deleting; Azure deletions are safe to retry. - If you stream logs from the worker, early deletion can truncate tail logs. If that matters, increase the grace period or set

stream_output

to false. Example robust delete in your cleanup flow:

from azure.identity import DefaultAzureCredential
from azure.mgmt.containerinstance import ContainerInstanceManagementClient
from azure.core.exceptions import ResourceNotFoundError

def safe_delete_container_group(subscription_id, resource_group, name):
    client = ContainerInstanceManagementClient(DefaultAzureCredential(), subscription_id)
    try:
        client.container_groups.begin_delete(resource_group, name).result()
    except ResourceNotFoundError:
        # Already deleted by the worker or another reaper
        pass

Best-practice pattern: - Tag your ACI groups from the work pool template (

"tags": {{ labels }}

) and only target those tags. - In the reaper, delete only: - Groups with your tag, AND - All containers in a terminal state, AND - Older than your grace period. Docs for reference: - Azure Container Instance worker - Customize job variables - Workers concept If you want, I can help you tune the grace-period logic to your workloads and add a quick schedule/automation to run the reaper.

You can absolutely drive cleanup off an Automation, but there isn’t a first-class “worker went offline” event to trigger on. The two reliable patterns are: A) Trigger when the work queue becomes NOT_READY (no worker polling) - Why: a queue becomes NOT_READY within ~60s if no worker is polling it, which correlates with your worker being down. - How (UI): - Create a small cleanup flow/deployment (e.g., your ACI reaper). - In Automations, create a new automation with a trigger on your specific Work Queue when it transitions to NOT_READY. - Add actions: - RunDeployment (your cleanup flow; pass params like subscription/resource group/pool/queue). - Optionally CallWebhook, SendNotification, etc. B) Proactive automation on the absence of queue polls (advanced) - Why: detect “no worker activity” over a time window. - How: Use a Proactive event trigger that expects queue poll events within X minutes and fires when none arrive. The exact event names vary; use the UI event picker to select the “work queue polled” event. - Example JSON shape (use the UI or REST API; fill in your IDs and use the event picker to choose the correct event name):

{
  "name": "Cleanup when no worker polling my-queue",
  "description": "Runs ACI cleanup when my-queue has no polls for 10m",
  "trigger": {
    "type": "event",
    "posture": "proactive",
    "expect": [
      "prefect.work-queue.polled"
    ],
    "within": 600,
    "match": {
      "prefect.resource.id": "prefect.work-queue:my-pool/my-queue"
    }
  },
  "actions": [
    {
      "type": "run-deployment",
      "source": "selected",
      "deployment_id": "<your-cleanup-deployment-id>",
      "parameters": {
        "__prefect_kind": "jinja",
        "subscription_id": "****",
        "resource_group": "****",
        "work_pool": "my-pool",
        "work_queue": "my-queue",
        "tag_key": "prefect-aci",
        "tag_value": "true",
        "min_age_minutes": 10
      }
    }
  ],
  "enabled": true
}

Notes: - Use the Automations UI “event picker” to select the exact event (names can differ by version). - Add a grace window (e.g., 5–10 minutes) so you don’t fire during short worker restarts. What to put in the cleanup deployment - Reuse the “reaper” flow we discussed. Target ACIs by tag (map Prefect labels to Azure tags via the ACI work pool template: add

"tags": {{ labels }}

to the container group resource), then delete only terminal/old-enough groups. - Keep this deployment running on a resilient worker (e.g., Process/Docker/VM) so it’s available when your ACI worker is down. If you prefer code-driven monitoring of “dead workers” - Another pattern: a scheduled “monitor” flow uses the Prefect client to detect workers whose last heartbeat is older than a threshold, then runs your cleanup tasks:

from datetime import datetime, timezone, timedelta
from prefect import flow
from prefect.client.orchestration import get_client

@flow
async def detect_dead_workers(max_age_seconds=600, work_pool_name: str | None = None):
    async with get_client() as client:
        workers = await client.read_workers()  # filter options available in client
        now = datetime.now(timezone.utc)
        dead = []
        for w in workers:
            if work_pool_name and w.work_pool_name != work_pool_name:
                continue
            if w.last_heartbeat_time and (now - w.last_heartbeat_time).total_seconds() > max_age_seconds:
                dead.append(w)
        # Call your cleanup routine here with pool/queue context...

Then schedule this flow or trigger it via a simple “time-based” automation. Docs - Automations overview: Creating automations - Event triggers: Event triggers - Templating parameters in automation actions: Access parameters in templates

- ACI worker reference: Azure Container Instance worker Want me to: - Wire up the work-queue NOT_READY automation for your specific pool/queue, or - Generate the cleanup deployment YAML with parameters so you can drop it in?

Yes. There’s no high-level Python SDK helper for this, but you can update a work pool’s base job template using the Prefect client against the REST API. Two practical approaches: Option 1: Use the CLI from Python (simple, recommended) - Generate your JSON in Python and shell out to the CLI:

import json
import subprocess
from pathlib import Path

pool_name = "my-aci-pool"

# Load or build your template dict in Python
template = {
    "job_configuration": {
        # ... copy the default and add:
        # e.g. add tags to the ARM template
        "arm_template": {
            # your ARM template here; ensure you include "tags": {{ labels }} on the container group
        }
    }
}

tmp = Path("aci_base_job_template.json")
tmp.write_text(json.dumps(template, indent=2))

subprocess.run(
    ["prefect", "work-pool", "update", pool_name, "--base-job-template", str(tmp)],
    check=True,
)

Option 2: Call the REST API with PrefectClient (no CLI, more code) - The REST endpoint supports PATCH /api/work_pools/{name} with a body containing

base_job_template

. - Example with the async client:

import asyncio
import json
from prefect.client.orchestration import get_client

async def update_pool_template(pool_name: str, template: dict):
    async with get_client() as client:
        # PATCH /api/work_pools/{name}
        resp = await client.request(
            "PATCH",
            f"/work_pools/{pool_name}",
            json={"base_job_template": template},
        )
        return resp

template = {
    "job_configuration": {
        # include the fields your pool type expects
        # and your ARM edits, e.g.:
        "arm_template": {
            # ...
            # inside the container group resource:
            # "tags": {{ labels }}
        }
    }
}

asyncio.run(update_pool_template("my-aci-pool", template))

Tips: - Start from the default template so you preserve required fields:

prefect work-pool get-default-base-job-template --type azure-container-instance > aci.json

Then load, modify, and send it back. - Validate your JSON before updating; if required fields are missing, the worker may fail to start jobs. Relevant docs: - Manage work pools - Update work pool (REST API) - ACI worker reference: Azure Container Instance worker If you share what you want to inject (e.g., adding

"tags": {{ labels }}

in the ARM template), I can sketch the minimal JSON patch you need.