Short answer: it has to be called on a Prefect Task, not a plain Python function. In Prefect 3.x,

.map()

is a method on

Task

objects. To use it with a plain function, wrap the function as a task (either with the

@task

decorator or by calling

task(fn)

). Examples:

from prefect import flow, task

@task
def add_one(x: int) -> int:
    return x + 1

@flow
def my_flow():
    futures = add_one.map([1, 2, 3])  # OK: .map on a Task
    return futures.result()

from prefect import flow, task

def multiply(x: int) -> int:
    return x * 2

@flow
def my_flow():
    futures = task(multiply).map([1, 2, 3])  # wrap function as a Task
    return futures.result()

If you prefer more manual control, you can also submit tasks individually:

from prefect import flow, task

@task
def work(x: int) -> int:
    return x * 2

@flow
def my_flow():
    futures = [work.submit(i) for i in [1, 2, 3]]
    return [f.result() for f in futures]

Yes — a task can return a tuple. Prefect treats the tuple as a single result value. A few practical notes: - Unpacking: You can unpack the tuple after resolving the future.

from prefect import flow, task

  @task
  def split(x: int) -> tuple[int, int]:
    return x, x * x

  @flow
  def my_flow():
    fut = split.submit(3)
    a, b = fut.result()  # (3, 9)
    return a, b

- Mapping with tuple outputs: Mapping doesn’t “fan out” tuple elements automatically. A downstream task will receive the whole tuple unless you split it yourself.

from prefect import flow, task

  @task
  def split(x: int) -> tuple[int, int]:
    return x, x * x

  @task
  def consume_pair(pair: tuple[int, int]) -> int:
    a, b = pair
    return a + b

  @flow
  def my_flow():
    pairs = split.map([1, 2, 3])        # PrefectFutureList of tuples
    sums = consume_pair.map(pairs)      # pass each tuple as a single arg
    return sums.result()                # [2, 6, 12]

- If you need to map a downstream task over the tuple elements as separate args, unzip first (ideally without blocking) using small helper tasks:

@task
  def left(p): return p[0]

  @task
  def right(p): return p[1]

  @task
  def add(a: int, b: int) -> int:
    return a + b

  @flow
  def my_flow():
    pairs = split.map([1, 2, 3])
    as_ = left.map(pairs)               # mapped extraction
    bs_ = right.map(pairs)
    sums = add.map(as_, bs_)            # map over two iterables
    return sums.result()                # [2, 6, 12]

- Persistence/serialization: Tuples are natively supported. If you’ve enabled result persistence (task-level or flow-level), the tuple will be serialized and stored. Otherwise it’s kept in-memory during the run. If you had a different use in mind (e.g., using tuple elements as separate upstreams without helper tasks), let me know and I can suggest a pattern that fits.

Yes. In Prefect 3.x there are several ways to kick off a flow from an external event. The best option depends on where you’re running (Cloud vs self-hosted) and what’s emitting the event. Common approaches: - Prefect Cloud webhooks (Cloud-only) - Create a webhook, then POST to its unique URL from your external system. The webhook turns the payload into a Prefect event. Use an Automation with an event trigger to run your deployment. - Example invocation:

curl -X POST <https://api.prefect.cloud/hooks/<your_webhook_id>> \
    -H "Content-Type: application/json" \
    -d '{"dataset_id":"sales-2024","note":"ready"}'

- Then set an Automation that expects your event and runs your deployment with parameters (e.g.,

dataset_id: "{{ event.resource.id }}"

). - Directly create a flow run via API (Cloud or self-hosted) - From any system that can make HTTP requests, call the deployment “create flow run” endpoint. - Cloud:

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": {"key": "value"}, "name": "external-trigger", "tags": ["external"]}'

- Self-hosted:

curl -X POST "http://<prefect-server>/api/deployments/<deployment_id>/create_flow_run" \
    -H "Content-Type: application/json" \
    -d '{"parameters": {"key": "value"}}'

- Python (anywhere you can use a Prefect client):

import asyncio
  from uuid import UUID
  from prefect.client.orchestration import get_client

  async def trigger():
      async with get_client() as client:
          fr = await client.deployments.create_flow_run_from_deployment(
              deployment_id=UUID("<deployment_id>"),
              parameters={"key": "value"},
              tags=["external"],
              name="external-trigger"
          )
          print(fr.id)

  asyncio.run(trigger())

- Emit custom events and use Automations (Cloud or self-hosted) - If your external system is Python-capable, emit a Prefect event and let an Automation react to it.

from prefect.events import emit_event

  emit_event(
      event="data.ready",
      resource={"prefect.resource.id": "dataset.sales-2024"},
      payload={"rows": 50000}
  )

- Configure an Automation with an event trigger to run your deployment when

data.ready

is received. - Deployment event triggers (define triggers on the deployment itself) - You can attach event triggers directly to a deployment so that matching events run it automatically.

from prefect import flow
  from prefect.events import DeploymentEventTrigger

  @flow
  def my_flow(dataset_id: str):
      ...

  my_flow.deploy(
      name="data-processor",
      work_pool_name="default",
      triggers=[
          DeploymentEventTrigger(
              enabled=True,
              expect=["data.ready"],
              match={"prefect.resource.id": "dataset.*"},
              parameters={"dataset_id": "{{ event.resource.id }}"},
          )
      ],
  )

Notes: - Webhooks are Prefect Cloud-only. Direct API calls, event-based Automations, and deployment triggers work in both Cloud and self-hosted. - External triggers usually target a deployment so a worker can pick up the run. If you can share where you’re running (Cloud or OSS) and what’s producing the event (HTTP webhook, message bus, Python app, etc.), I can recommend the simplest setup and provide exact steps.

Great question. In Prefect 3.x, the idiomatic way to gracefully handle per-item failures from a mapped task is to work with futures/states and avoid letting one failure abort the whole map. Here are the patterns that work well: - Get results without raising exceptions - Call

.result(raise_on_failure=False)

on each future. You’ll get either the value or an Exception you can log/skip.

futures = my_task.map(items)
  results = [f.result(raise_on_failure=False) for f in futures]
  successes = [r for r in results if not isinstance(r, Exception)]
  errors = [r for r in results if isinstance(r, Exception)]

- Inspect states explicitly - Return states instead of futures with

return_state=True

and branch on

state.is_completed()

/

state.is_failed()

.

states = my_task.map(items, return_state=True)
  successes = [s.data.result for s in states if s.is_completed()]
  failures = [s for s in states if s.is_failed()]

- Stream and handle as they complete - Use

as_completed

to process each item with try/except, in completion order.

from prefect.futures import as_completed
  futures = my_task.map(items)
  for fut in as_completed(futures):
      try:
          val = fut.result()
          # handle success
      except Exception as e:
          # handle failure

- Allow failures to flow downstream intentionally - If a downstream mapped task should handle failures item-by-item, wrap inputs with

allow_failure()

. The downstream task will receive either the value or the Exception.

from prefect.utilities.annotations import allow_failure

  upstream = fetch.map(ids)                 # some may fail
  downstream = handle_item.map(allow_failure(upstream))

- Retries and conditional retries - Set

retries

,

retry_delay_seconds

, and optionally

retry_condition_fn

on the task to automatically retry only certain failures.

@task(retries=2, retry_delay_seconds=5)
  def my_task(x): ...

- Failure hooks for side effects - Add

on_failure=[...]

via

.with_options()

to log/report each failed mapped run.

def log_failure(task, task_run, state):
      print(f"Failed: {task_run.id} - {state.message}")

  my_task2 = my_task.with_options(on_failure=[log_failure])
  futures = my_task2.map(items)

- Wait and then filter - If you want to ensure all are done first,

wait(futures)

, then pull results with

raise_on_failure=False

.

from prefect.futures import wait
  futures = my_task.map(items)
  done, _ = wait(futures)
  results = [f.result(raise_on_failure=False) for f in done]

When to use which: - Need a clean list of successes and errors at once:

.result(raise_on_failure=False)

or

return_state=True

. - Need streaming behavior:

as_completed

. - Need downstream logic to cope with failures per item:

allow_failure

. - Need automated resilience: task-level retries (+ optional

retry_condition_fn

). - Need side effects/alerts on failure:

on_failure

hook or Automations. If you share your specific use case (e.g., “skip bad records but keep good ones, then aggregate”), I can suggest a minimal pattern tailored to it.

Yes. A few good ways to get the inputs that produced a failure in a mapped task: - Keep inputs paired with futures (simple and reliable) -

.map

preserves order, so you can zip the original inputs with each future and inspect failures.

futures = my_task.map(items)
  for item, fut in zip(items, futures):
      result = fut.result(raise_on_failure=False)
      if isinstance(result, Exception):
          # You have the original input here
          print("Failed input:", item, "error:", result)

- Use an on_failure hook to read task parameters - State hooks receive the

task_run

object; you can read

task_run.parameters

(the bound args/kwargs).

def log_failure(task, task_run, state):
      print(f"{task.name} failed with params={task_run.parameters} message={state.message}")

  my_task_with_hook = my_task.with_options(on_failure=[log_failure])
  my_task_with_hook.map(items)

- Query the task run from the client using the future’s task_run_id - Useful if you’re handling failures in a flow and want exactly what was sent to the failing run.

from prefect.client.orchestration import get_client

  @flow
  async def f():
      futures = my_task.map(items)
      async with get_client() as client:
          for item, fut in zip(items, futures):
              r = fut.result(raise_on_failure=False)
              if isinstance(r, Exception):
                  tr = await client.read_task_run(fut.task_run_id)
                  print("Failed params:", tr.parameters)

- Pass failures and original inputs to a downstream handler with allow_failure - Your handler gets both the exception (if any) and the input that caused it.

from prefect.utilities.annotations import allow_failure

  upstream = my_task.map(items)
  handled = handle_error.map(allow_failure(upstream), items)

- Inside the task itself - If you want to log inputs from within the task at failure time, grab them from the run context.

from prefect.context import get_run_context
  from prefect import task

  @task
  def my_task(a, b):
      params = get_run_context().parameters  # {'a': ..., 'b': ...}
      ...

Pick the approach that fits where you’re handling the failure: - In the flow: zip inputs to futures, or read

task_run.parameters

via the client. - At failure time automatically: use an

on_failure

hook. - In a downstream “error handler” task: use

allow_failure

and pass the original args alongside.