list_task_instance_tries
Retrieve execution attempts for a specific Airflow task instance to monitor performance and debug failures. Use DAG ID, DAG run ID, and task ID to query task execution history.
Instructions
List task instance tries by DAG ID, DAG run ID, and task ID
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| dag_id | Yes | ||
| dag_run_id | Yes | ||
| task_id | Yes | ||
| limit | No | ||
| offset | No | ||
| order_by | No |
Implementation Reference
- src/airflow/taskinstance.py:129-149 (handler)The async handler function for the 'list_task_instance_tries' tool. It accepts dag_id, dag_run_id, task_id, optional limit, offset, order_by; calls the Airflow TaskInstanceApi.get_task_instance_tries and returns the response as TextContent.
async def list_task_instance_tries( dag_id: str, dag_run_id: str, task_id: str, limit: Optional[int] = None, offset: Optional[int] = None, order_by: Optional[str] = None, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: # Build parameters dictionary kwargs: Dict[str, Any] = {} if limit is not None: kwargs["limit"] = limit if offset is not None: kwargs["offset"] = offset if order_by is not None: kwargs["order_by"] = order_by response = task_instance_api.get_task_instance_tries( dag_id=dag_id, dag_run_id=dag_run_id, task_id=task_id, **kwargs ) return [types.TextContent(type="text", text=str(response.to_dict()))] - src/airflow/taskinstance.py:28-33 (registration)Registration tuple for the 'list_task_instance_tries' tool in the get_all_functions() list, including the handler function, name, description, and read-only flag.
( list_task_instance_tries, "list_task_instance_tries", "List task instance tries by DAG ID, DAG run ID, and task ID", True, ),