a json artifact

variables serve this purpose

but if you're just trying to see what you're passing to a callable, then a spy pattern using pytest/unittest is common (ie you don't have to write to the api if you're writing unit tests)

Totally—here are a few quick, reliable “spy” patterns for asserting Prefect task call params in Prefect 3.x. They all work locally and with the engine; pick the one that best fits your test style. Pattern A: Patch the task’s underlying function with a MagicMock spy - Prefect executes the underlying Python function stored on the task; spying on that function captures the actual args/kwargs used by the engine. - Works great with threads; if you use processes, patch the symbol in the module under test so it’s visible in the child process.

# my_flow.py
from prefect import flow, task

@task
def add(x, y=2):
    return x + y

@flow
def my_flow():
    return add(1, y=3)

# test_my_flow.py
from unittest.mock import MagicMock
import inspect
import my_flow

def test_add_called_with_expected_params():
    # capture the original signature to normalize args->param names
    orig_sig = inspect.signature(my_flow.add.fn)

    spy = MagicMock(wraps=my_flow.add.fn)
    my_flow.add.fn = spy  # patch the function that the task actually executes

    my_flow.my_flow()

    spy.assert_called_once()
    ba = orig_sig.bind_partial(*spy.call_args.args, **spy.call_args.kwargs)
    assert ba.arguments == {"x": 1, "y": 3}

Pattern B: Wrap the task function to record bound parameters via TaskRunContext - This gives you the canonical parameters dict Prefect computed (no need to re-bind args).

# conftest.py or inline in test
import functools, inspect
from prefect.context import get_run_context

def spy_on_task(task_obj):
    calls = []
    original = task_obj.fn

    if inspect.iscoroutinefunction(original):
        @functools.wraps(original)
        async def wrapper(*a, **k):
            params = get_run_context().parameters.copy()
            calls.append(params)
            return await original(*a, **k)
    else:
        @functools.wraps(original)
        def wrapper(*a, **k):
            params = get_run_context().parameters.copy()
            calls.append(params)
            return original(*a, **k)

    task_obj.fn = wrapper
    return calls

Usage:

# test_my_flow.py
import my_flow
from .conftest import spy_on_task

def test_params_with_context_capture():
    calls = spy_on_task(my_flow.add)
    my_flow.my_flow()
    assert calls == [{"x": 1, "y": 3}]

Notes and gotchas - Retries: Your spy will capture each retry as an additional call; assert on call count as needed. - ProcessPoolTaskRunner: If you use processes, ensure the patched/wrapped task is defined in the module that child processes import (i.e., patch the symbol in the module under test before the pool starts). - Keep task config intact: These approaches only wrap/replace the underlying function; they do not change task options like retries, timeouts, etc. Related docs - Tasks: Prefect tasks - Context: get_run_context API - Testing guide: Testing flows and tasks If you share a snippet of your flow/task setup, I can tailor the spy wrapper to your exact structure (async/sync, mapping, etc.).