<@ULVA73B9P> In Prefect 3, is it possible to manua...
# ask-marvin
n
@Marvin In Prefect 3, is it possible to manually trigger a custom event via API, and start a flow when that event is triggered? the name and parameters of the flow must be contained in the event payload. Basically, the event decides what deployed flow must be started
m
thought for 695.4 seconds
Short answer: Yes—Prefect 3 supports posting custom events and using an Automation to start a flow run when those events arrive. Parameters can be mapped from the event payload. However, dynamically choosing which deployment to run purely from a payload “name” is not built-in—you either need to include the deployment UUID as a related resource in the event (recommended), or route through a small webhook/router that looks up the deployment by name and runs it. Here are the workable patterns. Pattern A (recommended): Event embeds the target deployment UUID; Automation infers it - Emit your custom event and include the target deployment as a related resource with id format prefect.deployment.<DEPLOYMENT_UUID>. - Create one Automation with a RunDeployment action using source="inferred" so it picks the deployment from the event. - Map all flow parameters from event.payload using Jinja. Emit event (Prefect Cloud):
Copy code
curl -X POST \
  "<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>/events>" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '[
    {
      "event": "custom.webhook.received",
      "resource": { "prefect.resource.id": "webhook.request.12345" },
      "related": [
        {
          "prefect.resource.id": "prefect.deployment.be777bbd-4b15-49f3-bc1f-4d109374cee2",
          "prefect.resource.role": "deployment",
          "prefect.resource.name": "target-flow/target-deployment"
        }
      ],
      "payload": {
        "param_a": "foo",
        "param_b": 42
      }
    }
  ]'
Notes: - The body must be a JSON array of events (even for one event). - For self-hosted: POST http//&lt;server&gt;8080/api/events - Related resource id must be prefect.deployment.<UUID> Automation (single automation handles any deployment referenced by the event):
Copy code
from prefect.events import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import RunDeployment

automation = Automation(
    name="Run deployments from custom events",
    trigger=EventTrigger(expect={"custom.webhook.received"}),
    actions=[
        RunDeployment(
            source="inferred",  # infer deployment_id from the event’s related resources
            parameters={
                "param_a": {"__prefect_kind": "jinja", "template": "{{ event.payload.param_a }}"},
                "param_b": {"__prefect_kind": "jinja", "template": "{{ event.payload.param_b }}"}
            }
        )
    ],
    enabled=True,
)
# Create via API or configure equivalent in the UI
Pattern B: Only a deployment name is available in payload RunDeployment.deployment_id cannot be templated and Prefect cannot dynamically choose a deployment by name from the payload inside the same action. Use one of these: - B1: Webhook router - Use a CallWebhook action that posts the event data (including the desired deployment name) to a small service (e.g., AWS Lambda). - In that service, use Prefect’s API/SDK to look up the deployment by name and run it with parameters from the payload. Router example (Python service):
Copy code
from prefect.client.orchestration import get_client

def run_by_name(flow_and_deployment: str, params: dict):
    flow_name, dep_name = flow_and_deployment.split("/", 1)
    with get_client(sync_client=True) as client:
        dep = client.read_deployment_by_name(f"{flow_name}/{dep_name}")
        return client.create_flow_run_from_deployment(
            deployment_id=dep.id,
            parameters=params,
            tags=["event-driven"]
        )
- B2: Router flow - Create one “router” deployment that is triggered by the event (fixed deployment_id). - Pass the entire event payload into this router flow; inside the flow, use the SDK to look up the target deployment by name and call run_deployment/read_deployment_by_name. - This keeps everything inside Prefect without a separate webhook service. Router flow: ```
from prefect import flow from prefect.deployments import run_deployment @flow def router(payload: dict): target = payload["deployment_name"] # e.g. "my-flow/my-deployment" params = payload.get("parameters", {}) run_deployment(name=target, parameters=params, timeout=0)
Copy code
Deploy router with an Automation:
RunDeployment( source="selected", deployment_id=<ROUTER_DEPLOYMENT_UUID>, parameters={ "payload": { "__prefect_kind": "json", "value": {"__prefect_kind": "jinja", "template": "{{ event.payload | tojson }}"} } } ) ``` - B3: Small fixed set of targets - Create multiple automations (or multiple actions) each with a fixed deployment_id and mutually exclusive match conditions (based on resource labels you include in the event). This scales poorly if you have many target deployments. Important details and constraints - Posting events: - Endpoint: POST /api/events (Cloud: https://api.prefect.cloud/api/accounts/{ACCOUNT_ID}/workspaces/{WORKSPACE_ID}/events) - Body must be a list of events - Include Authorization: Bearer <API key> for Cloud - Parameter mapping: - Use Jinja in action.parameters to pull values from event.payload - Example: {"param": {"__prefect_kind": "jinja", "template": "{{ event.payload.x }}"}} - Choosing the deployment: - RunDeployment supports source="selected" (fixed deployment_id) or source="inferred" (deployment pulled from event’s related resources) - deployment_id is not templatable - Event resource ids: - Deployment related resource id format: prefect.deployment.<UUID> - Security and limits: - Keep payload sizes modest (Cloud enforces limits) - Authenticate requests to Cloud with a workspace API key Docs - Events: Events - Automations (+ Jinja templating): Automations - Cloud API docs (Events): Cloud API reference A couple of clarifying questions to get you the best setup: - Can your event emitter resolve the deployment UUID up front? If yes, Pattern A is the cleanest: add it as a related resource and use source="inferred". - If you only have a “flow/deployment” name, do you prefer a tiny webhook/router (B1) or a Prefect “router” flow (B2)?