get_dag
Retrieve comprehensive details of a specific DAG in Apache Airflow, including schedule, owners, tags, and description, to simplify cluster monitoring and management.
Instructions
[Tool Role]: Retrieves detailed information for a specific DAG.
Args: dag_id: The DAG ID to get details for
Returns: Comprehensive DAG details: dag_id, schedule_interval, start_date, owners, tags, description, etc.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| dag_id | Yes |
Implementation Reference
- The MCP tool handler for 'get_dag'. This is the entrypoint decorated with @mcp.tool() that handles the tool invocation and delegates to the detailed info helper.@mcp.tool() async def get_dag(dag_id: str) -> Dict[str, Any]: """ [Tool Role]: Retrieves detailed information for a specific DAG. Args: dag_id: The DAG ID to get details for Returns: Comprehensive DAG details """ return await get_dag_detailed_info(dag_id)
- Core helper function that performs the actual Airflow API request to fetch detailed DAG information and formats the response.async def get_dag_detailed_info(dag_id: str) -> Dict[str, Any]: """ Internal helper function to get detailed DAG information. This function contains the actual implementation logic that can be reused. """ if not dag_id: raise ValueError("dag_id must not be empty") resp = await airflow_request("GET", f"/dags/{dag_id}") resp.raise_for_status() dag = resp.json() return { "dag_id": dag.get("dag_id"), "dag_display_name": dag.get("dag_display_name"), "description": dag.get("description"), "schedule_interval": dag.get("schedule_interval"), "start_date": dag.get("start_date"), "end_date": dag.get("end_date"), "is_active": dag.get("is_active"), "is_paused": dag.get("is_paused"), "owners": dag.get("owners"), "tags": [t.get("name") for t in dag.get("tags", [])], "catchup": dag.get("catchup"), "max_active_runs": dag.get("max_active_runs"), "max_active_tasks": dag.get("max_active_tasks"), "has_task_concurrency_limits": dag.get("has_task_concurrency_limits"), "has_import_errors": dag.get("has_import_errors"), "next_dagrun": dag.get("next_dagrun"), "next_dagrun_data_interval_start": dag.get("next_dagrun_data_interval_start"), "next_dagrun_data_interval_end": dag.get("next_dagrun_data_interval_end") }
- src/mcp_airflow_api/tools/v1_tools.py:12-27 (registration)Registration entrypoint for v1 API version that sets the airflow_request function and calls register_common_tools which defines and registers the get_dag handler.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:13-25 (registration)Registration entrypoint for v2 API version that sets the airflow_request function and calls register_common_tools which defines and registers the get_dag handler.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)
- The function signature and docstring provide the input schema (dag_id: str) and output type (Dict[str, Any]), used by MCP for tool schema generation.@mcp.tool() async def get_dag(dag_id: str) -> Dict[str, Any]: """ [Tool Role]: Retrieves detailed information for a specific DAG. Args: dag_id: The DAG ID to get details for Returns: Comprehensive DAG details """ return await get_dag_detailed_info(dag_id)