all_dag_import_summary

Identify import errors across all DAGs in Apache Airflow to diagnose and resolve deployment issues, enabling reliable workflow execution.

Instructions

[Tool Role]: Provides summary of import errors across all DAGs.

Input Schema

TableJSON Schema

Name	Required	Description	Default
No arguments

Implementation Reference

src/mcp_airflow_api/tools/common_tools.py:419-441 (handler)

The main execution handler for the 'all_dag_import_summary' MCP tool. Queries Airflow's importErrors API endpoint (limit=1000), groups errors by filename, truncates long stacktraces, and returns structured summary with totals.

@mcp.tool()
async def all_dag_import_summary() -> Dict[str, Any]:
    """[Tool Role]: Provides summary of import errors across all DAGs."""
    resp = await airflow_request("GET", "/importErrors?limit=1000")
    resp.raise_for_status()
    data = resp.json()
    
    import_summary = {}
    for error in data.get("import_errors", []):
        filename = error.get("filename", "unknown")
        if filename not in import_summary:
            import_summary[filename] = []
        import_summary[filename].append({
            "id": error.get("id"),
            "timestamp": error.get("timestamp"),
            "stacktrace": error.get("stacktrace", "")[:200] + "..."
        })
    
    return {
        "import_errors_summary": import_summary,
        "total_errors": len(data.get("import_errors", [])),
        "files_with_errors": len(import_summary)
    }

src/mcp_airflow_api/tools/v1_tools.py:13-27 (registration)

v1 API version registration entry point. Sets v1-specific airflow_request function and calls register_common_tools(mcp) which registers the 'all_dag_import_summary' tool via its @mcp.tool() decorator.

def register_tools(mcp):
    """Register v1 tools by importing common tools with v1 request function."""
    
    logger.info("Initializing MCP server for Airflow API v1")
    logger.info("Loading Airflow API v1 tools (Airflow 2.x)")
    
    # Set the global request function to v1
    common_tools.airflow_request = airflow_request_v1
    
    # Register all 56 common tools (includes management tools)
    common_tools.register_common_tools(mcp)
    
    # V1 has no exclusive tools - all tools are shared with v2
    
    logger.info("Registered all Airflow API v1 tools (56 tools: 43 core + 13 management tools)")

src/mcp_airflow_api/tools/v2_tools.py:14-25 (registration)

v2 API version registration entry point. Sets v2-specific airflow_request function and calls register_common_tools(mcp) which registers the 'all_dag_import_summary' tool via its @mcp.tool() decorator.

def register_tools(mcp):
    """Register v2 tools: common tools + v2-exclusive asset tools."""
    
    logger.info("Initializing MCP server for Airflow API v2")
    logger.info("Loading Airflow API v2 tools (Airflow 3.0+)")
    
    # Set the global request function to v2
    common_tools.airflow_request = airflow_request_v2
    
    # Register all 43 common tools
    common_tools.register_common_tools(mcp)

src/mcp_airflow_api/mcp_main.py:160-173 (helper)

Documentation in troubleshooting prompt template recommending the use of 'all_dag_import_summary' tool for import error analysis.

**Import Errors**: Use `list_import_errors` and `all_dag_import_summary`

### 3. Deep Dive Investigation
- Examine task logs: `get_task_instance_logs`
- Review task details: `get_task_instance_details`
- Check XCom data: `list_xcom_entries`
- Analyze event logs: `list_event_logs`

### 4. Monitoring and Prevention
- Set up regular health checks
- Monitor resource utilization trends
- Review configuration settings
"""

src/mcp_airflow_api/mcp_main.py:383-390 (registration)

Main MCP server initialization calls the appropriate v1_tools.register_tools(mcp) or v2_tools.register_tools(mcp) based on detected Airflow API version, which ultimately registers the tool.

    logger.info("Loading Airflow API v1 tools (Airflow 2.x)")
    from mcp_airflow_api.tools import v1_tools
    v1_tools.register_tools(mcp)
elif api_version == "v2":
    logger.info("Loading Airflow API v2 tools (Airflow 3.0+)")
    from mcp_airflow_api.tools import v2_tools
    v2_tools.register_tools(mcp)

MCP-Airflow-API