get_dag_details

Instructions

Implementation Reference

• src/astro_airflow_mcp/server.py:360-374 (handler)

  Internal implementation (handler) for get_dag_details tool. Calls adapter.get_dag(dag_id) and returns JSON.

  def _get_dag_details_impl(dag_id: str) -> str:
      """Internal implementation for getting details about a specific DAG.
  
      Args:
          dag_id: The ID of the DAG to get details for
  
      Returns:
          JSON string containing the DAG details
      """
      try:
          adapter = _get_adapter()
          data = adapter.get_dag(dag_id)
          return json.dumps(data, indent=2)
      except Exception as e:
          return str(e)

• src/astro_airflow_mcp/server.py:377-412 (schema)

  MCP tool registration and schema definition for get_dag_details. Decorated with @mcp.tool() and includes docstring describing input/output.

  @mcp.tool()
  def get_dag_details(dag_id: str) -> str:
      """Get detailed information about a specific Apache Airflow DAG.
  
      Use this tool when the user asks about:
      - "Show me details for DAG X" or "What are the details of DAG Y?"
      - "Tell me about DAG Z" or "Get information for this specific DAG"
      - "What's the schedule for DAG X?" or "When does this DAG run?"
      - "Is DAG Y paused?" or "Show me the configuration of DAG Z"
      - "Who owns this DAG?" or "What are the tags for this workflow?"
  
      Returns complete DAG information including:
      - dag_id: Unique identifier for the DAG
      - is_paused: Whether the DAG is currently paused
      - is_active: Whether the DAG is active
      - is_subdag: Whether this is a SubDAG
      - fileloc: File path where the DAG is defined
      - file_token: Unique token for the DAG file
      - owners: List of DAG owners
      - description: Human-readable description of what the DAG does
      - schedule_interval: Cron expression or timedelta for scheduling
      - tags: List of tags/labels for categorization
      - max_active_runs: Maximum number of concurrent runs
      - max_active_tasks: Maximum number of concurrent tasks
      - has_task_concurrency_limits: Whether task concurrency limits are set
      - has_import_errors: Whether the DAG has import errors
      - next_dagrun: When the next DAG run is scheduled
      - next_dagrun_create_after: Earliest time for next DAG run creation
  
      Args:
          dag_id: The ID of the DAG to get details for
  
      Returns:
          JSON with complete details about the specified DAG
      """
      return _get_dag_details_impl(dag_id=dag_id)

• src/astro_airflow_mcp/server.py:377-412 (registration)

  The tool is registered via the @mcp.tool() decorator on line 377 on the FastMCP instance 'mcp' (defined on line 179).

  @mcp.tool()
  def get_dag_details(dag_id: str) -> str:
      """Get detailed information about a specific Apache Airflow DAG.
  
      Use this tool when the user asks about:
      - "Show me details for DAG X" or "What are the details of DAG Y?"
      - "Tell me about DAG Z" or "Get information for this specific DAG"
      - "What's the schedule for DAG X?" or "When does this DAG run?"
      - "Is DAG Y paused?" or "Show me the configuration of DAG Z"
      - "Who owns this DAG?" or "What are the tags for this workflow?"
  
      Returns complete DAG information including:
      - dag_id: Unique identifier for the DAG
      - is_paused: Whether the DAG is currently paused
      - is_active: Whether the DAG is active
      - is_subdag: Whether this is a SubDAG
      - fileloc: File path where the DAG is defined
      - file_token: Unique token for the DAG file
      - owners: List of DAG owners
      - description: Human-readable description of what the DAG does
      - schedule_interval: Cron expression or timedelta for scheduling
      - tags: List of tags/labels for categorization
      - max_active_runs: Maximum number of concurrent runs
      - max_active_tasks: Maximum number of concurrent tasks
      - has_task_concurrency_limits: Whether task concurrency limits are set
      - has_import_errors: Whether the DAG has import errors
      - next_dagrun: When the next DAG run is scheduled
      - next_dagrun_create_after: Earliest time for next DAG run creation
  
      Args:
          dag_id: The ID of the DAG to get details for
  
      Returns:
          JSON with complete details about the specified DAG
      """
      return _get_dag_details_impl(dag_id=dag_id)

• src/astro_airflow_mcp/adapters/airflow_v3.py:86-89 (helper)

  AirflowV3Adapter.get_dag() calls Airflow REST API endpoint dags/{dag_id}

  def get_dag(self, dag_id: str) -> dict[str, Any]:
      """Get details of a specific DAG."""
      return self._call(f"dags/{dag_id}")

• src/astro_airflow_mcp/adapters/airflow_v2.py:20-22 (helper)

  AirflowV2Adapter.get_dag() calls Airflow REST API endpoint dags/{dag_id}

  def get_dag(self, dag_id: str) -> dict[str, Any]:
      """Get details of a specific DAG."""
      return self._call(f"dags/{dag_id}")