list_task_instances_batch
Retrieve multiple Airflow task instances by filtering with date ranges and state criteria to monitor workflow execution.
Instructions
[Tool Role]: Lists task instances in batch with date and state filtering.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | ||
| offset | No | ||
| start_date_gte | No | ||
| start_date_lte | No | ||
| state | No |
Implementation Reference
- The core handler function decorated with @mcp.tool() that executes the list_task_instances_batch tool. It queries the Airflow API /taskInstances endpoint with pagination, date range, and state filters, then enriches the response with a state summary.@mcp.tool() async def list_task_instances_batch( limit: int = 100, offset: int = 0, start_date_gte: Optional[str] = None, start_date_lte: Optional[str] = None, state: Optional[List[str]] = None ) -> Dict[str, Any]: """[Tool Role]: Lists task instances in batch with date and state filtering.""" params = {'limit': limit, 'offset': offset} if start_date_gte: params['start_date_gte'] = start_date_gte if start_date_lte: params['start_date_lte'] = start_date_lte if state: params['state'] = state query_string = "&".join([f"{k}={v}" for k, v in params.items()]) resp = await airflow_request("GET", f"/taskInstances?{query_string}") resp.raise_for_status() data = resp.json() # Add summary statistics task_instances = data.get("task_instances", []) state_summary = {} for task in task_instances: task_state = task.get("state", "unknown") state_summary[task_state] = state_summary.get(task_state, 0) + 1 data["state_summary"] = state_summary return data
- src/mcp_airflow_api/tools/v1_tools.py:23-23 (registration)Calls register_common_tools(mcp) which registers the list_task_instances_batch tool (among other common tools) for Airflow API v1 usage.common_tools.register_common_tools(mcp)
- src/mcp_airflow_api/tools/v2_tools.py:24-24 (registration)Calls register_common_tools(mcp) which registers the list_task_instances_batch tool (among other common tools) for Airflow API v2 usage.common_tools.register_common_tools(mcp)