Short answer: use the runtime helper to read the task’s inputs, then rehydrate your Pydantic model. In Prefect 3.x, task hooks (e.g.,

on_completion

) run in the same process as the task and you can read the effective inputs for the current task run via

prefect.runtime.task_run.parameters()

. That returns a dict of the call args (JSON-serializable), so if your task took a Pydantic model, you’ll get a dict you can validate back into your model. Example with a task-level on_completion hook:

from prefect import task
from prefect.runtime.task_run import parameters
from pydantic import BaseModel

# Your Pydantic model
class MyModel(BaseModel):
    id: int
    name: str

# Task hook signature: (task, task_run, state)
def on_done(task, task_run, state):
    # Get the actual inputs used for this task run
    params = parameters()  # -> dict[str, Any]
    raw_model = params.get("my_model")

    # Rehydrate the model
    # Pydantic v2:
    model = MyModel.model_validate(raw_model)
    # If you're on Pydantic v1, use:
    # model = MyModel.parse_obj(raw_model)

    # ...do whatever you need with `model` here...
    print(f"Completed with model: {model}")

@task(on_completion=[on_done])
def my_task(my_model: MyModel):
    # task logic
    return "ok"

Notes: - Hook callable signature for tasks is

(task, task_run, state)

. You can use

task_run.id

,

task_run.name

, etc., but the simplest way to get inputs is

prefect.runtime.task_run.parameters()

. - The same pattern works in async hooks; just make your hook

async def

and await anything you need. - If you’re on Pydantic v1 instead of v2, switch

model_validate

to

parse_obj

. If you’re on a different Pydantic version or have a different parameter name than

my_model

, let me know and I’ll tailor the example.