get_task_runs_by_flow_run
Retrieve task runs for a specific flow run in Prefect, with options to filter by state type, limit results, and paginate through large datasets.
Instructions
Get task runs for a specific flow run.
Args: flow_run_id: The flow run UUID limit: Maximum number of task runs to return offset: Number of task runs to skip state_type: Filter by state type (e.g., "RUNNING", "COMPLETED", "FAILED")
Returns: A list of task runs for the specified flow run
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| flow_run_id | Yes | ||
| limit | No | ||
| offset | No | ||
| state_type | No |
Implementation Reference
- src/mcp_prefect/task_run.py:154-205 (handler)The core handler function decorated with @mcp.tool, which registers and implements the get_task_runs_by_flow_run tool. It retrieves task runs associated with a specific flow run using the Prefect client, supports filtering by state_type, limit, and offset, adds UI links, and returns formatted results.
@mcp.tool async def get_task_runs_by_flow_run( flow_run_id: str, limit: Optional[int] = None, offset: Optional[int] = None, state_type: Optional[str] = None, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: """ Get task runs for a specific flow run. Args: flow_run_id: The flow run UUID limit: Maximum number of task runs to return offset: Number of task runs to skip state_type: Filter by state type (e.g., "RUNNING", "COMPLETED", "FAILED") Returns: A list of task runs for the specified flow run """ async with get_client() as client: # Build filter using new filter objects flow_run_filter = FlowRunFilter( id=FlowRunFilterId(any_=[UUID(flow_run_id)]) ) task_run_filter = None if state_type: task_run_filter = TaskRunFilter( state=TaskRunFilterState( type=TaskRunFilterStateType(any_=[state_type.upper()]) ) ) task_runs = await client.read_task_runs( flow_run_filter=flow_run_filter, task_run_filter=task_run_filter, limit=limit, offset=offset or 0 ) # Add UI links to each task run task_runs_result = { "task_runs": [ { **task_run.model_dump(), "ui_url": get_task_run_url(str(task_run.id)) } for task_run in task_runs ] } return [types.TextContent(type="text", text=str(task_runs_result))]