MCP-Airflow-API

Instructions

Implementation Reference

• src/mcp_airflow_api/tools/common_tools.py:752-783 (handler)

  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)