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
- 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)
- 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)