MCP-Airflow-API

Instructions

Implementation Reference

• src/mcp_airflow_api/tools/common_tools.py:75-86 (handler)

  The handler function for the 'get_dag' MCP tool. Decorated with @mcp.tool() for automatic registration. Delegates core logic to the helper function get_dag_detailed_info.

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

• src/mcp_airflow_api/functions.py:416-445 (helper)

  Core helper function implementing the DAG details retrieval via Airflow REST API (/dags/{dag_id}), parses and formats the response into a structured dictionary.

  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:23-23 (registration)

  Registration call for common tools (including get_dag) in the v1 tools module.

  common_tools.register_common_tools(mcp)

• src/mcp_airflow_api/tools/v2_tools.py:24-24 (registration)

  Registration call for common tools (including get_dag) in the v2 tools module.

  common_tools.register_common_tools(mcp)