Great question! In Prefect 2.x there are a few good ways to pass (or “mock”) an API request payload into a flow at launch time. Which one fits best depends on how you’re triggering runs and whether you’re testing or running in prod. Quick options: - Programmatically trigger a deployment with parameters (best if you own the API) - Prefect Cloud Webhook + Automation to map request → flow params (best for external webhooks) - CLI parameters for manual testing - Pause at start and send input later (human- or system-in-the-loop) - Mock external HTTP calls in tests Below are minimal examples for each. 1) Programmatically trigger with parameters (API or backend service) Make your flow accept parameters:

from prefect import flow

@flow
def my_flow(payload: dict):
    # use payload here
    ...

From Python (e.g., your API server), create a flow run with the payload:

import asyncio
from prefect import get_client

async def launch(payload: dict, deployment_id: str):
    async with get_client() as client:
        fr = await client.create_flow_run_from_deployment(
            deployment_id=deployment_id,
            parameters={"payload": payload},
            name="api-triggered"
        )
    return fr

Or via HTTP to Prefect Cloud (replace account/workspace/deployment IDs and API key):

curl -X POST "<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>/deployments/<DEPLOYMENT_ID>/create_flow_run>" \
  -H "Authorization: Bearer <PREFECT_API_KEY>" \
  -H "Content-Type: application/json" \
  -d '{"parameters":{"payload": {"id":123,"event":"update"}},"name":"api-triggered"}'

2) Prefect Cloud Webhook + Automation (for external systems) - Create a Cloud Webhook in the UI that captures the incoming HTTP request (JSON body, headers, etc.). - Build an Automation that triggers on that webhook event and runs your deployment. - In the Automation action, set deployment parameters using the event payload (e.g., map the incoming JSON body to your flow’s

payload

param). This lets you go straight from an external webhook → flow run with the request body as parameters. 3) CLI for manual testing If you just want to simulate a run with payload:

prefect deployment run "my-flow/my-deployment" -p payload='{"id":123,"event":"update"}' --watch

Notes: - Ensure your flow signature matches the parameter name, e.g.,

@flow def my_flow(payload: dict): ...

- Values are parsed as JSON when possible; quote appropriately. 4) Pause for input at start (optional, for interactive or delayed injection) If you want the flow to start, pause, then receive the payload after:

from prefect import flow, pause_flow_run
from prefect.input import RunInput

class Payload(RunInput):
    id: int
    event: str

@flow
async def my_flow_waits():
    payload = await pause_flow_run(wait_for_input=Payload)
    # continue with payload

Send the input later (from code with the flow_run_id):

from prefect.input import send_input

await send_input(
    run_input={"id": 123, "event": "update"},
    flow_run_id="<FLOW_RUN_ID>",
    key_prefix="payload"
)

This is useful for human-in-the-loop, approvals, or when data arrives after scheduling. 5) Mock external HTTP calls during tests (so you can simulate API behavior) If your flow calls external APIs, mock them in tests so you don’t hit the real services: ``` import pytest import httpx from prefect.testing.utilities import AsyncMock, prefect_test_harness @pytest.fixture(autouse=True, scope="session") def _prefect_harness(): with prefect_test_harness(): yield @pytest.mark.asyncio async def test_flow_with_mock(monkeypatch): # Suppose your flow uses httpx.AsyncClient.get(...) mock_get = AsyncMock(return_value=httpx.Response( status_code=200, json={"ok": True, "data": {"id": 123}} )) monkeypatch.setattr("httpx.AsyncClient.get", mock_get) # Call your flow with a test payload from my_module import my_flow

result = my_flow({"id": 123}) mock_get.assert_called_once() # Assert on result as needed ``` Which route do you want? - If you own the API and can call Prefect: use “create_flow_run_from_deployment” with

parameters

(Option 1). - If an external system must call into Prefect: use Cloud Webhooks + Automation (Option 2). - If you’re just testing manually: use the CLI (Option 3). - If you want to inject data after scheduling: use RunInput (Option 4). - If you mean “mock” as in “don’t hit external APIs during tests”: use the AsyncMock pattern (Option 5). Can you share how you’re triggering runs today (own API vs external webhook vs manual) and whether you’re trying to simulate test inputs or wire real payloads into production runs? I can tailor a drop-in example for your setup.