get_dag_source

Instructions

Implementation Reference

• src/astro_airflow_mcp/server.py:483-503 (registration)

  MCP tool registration for 'get_dag_source' via @mcp.tool() decorator. This is where the tool is exposed to MCP clients. The decorated function calls _get_dag_source_impl().

  @mcp.tool()
  def get_dag_source(dag_id: str) -> str:
      """Get the source code for a specific Apache Airflow DAG.
  
      Use this tool when the user asks about:
      - "Show me the code for DAG X" or "What's the source of DAG Y?"
      - "How is DAG Z implemented?" or "What does the DAG file look like?"
      - "Can I see the Python code for this workflow?"
      - "What tasks are defined in the DAG code?"
  
      Returns the DAG source file contents including:
      - content: The actual Python source code of the DAG file
      - file_token: Unique identifier for the source file
  
      Args:
          dag_id: The ID of the DAG to get source code for
  
      Returns:
          JSON with DAG source code and metadata
      """
      return _get_dag_source_impl(dag_id=dag_id)

• src/astro_airflow_mcp/server.py:466-480 (handler)

  Internal implementation _get_dag_source_impl() that calls adapter.get_dag_source(dag_id) and returns JSON. This is the core handler logic that bridges MCP tools to the Airflow adapter.

  def _get_dag_source_impl(dag_id: str) -> str:
      """Internal implementation for getting DAG source code from Airflow.
  
      Args:
          dag_id: The ID of the DAG to get source code for
  
      Returns:
          JSON string containing the DAG source code and metadata
      """
      try:
          adapter = _get_adapter()
          source_data = adapter.get_dag_source(dag_id)
          return json.dumps(source_data, indent=2)
      except Exception as e:
          return str(e)

• src/astro_airflow_mcp/adapters/airflow_v3.py:90-95 (helper)

  Airflow 3.x adapter implementation: calls REST API endpoint 'dagSources/{dag_id}' directly since Airflow 3 uses dag_id for source lookup.

  def get_dag_source(self, dag_id: str) -> dict[str, Any]:
      """Get source code of a DAG.
  
      Note: Airflow 3 directly uses dag_id for source lookup.
      """
      return self._call(f"dagSources/{dag_id}")

• src/astro_airflow_mcp/adapters/airflow_v2.py:24-36 (helper)

  Airflow 2.x adapter implementation: first fetches DAG details to get the file_token, then calls 'dagSources/{file_token}' since Airflow 2 requires file_token for source lookup.

  def get_dag_source(self, dag_id: str) -> dict[str, Any]:
      """Get source code of a DAG.
  
      Note: Airflow 2 requires a file_token, so we get DAG details first.
      """
      # First, get the DAG to get its file_token
      dag_data = self.get_dag(dag_id)
      file_token = dag_data.get("file_token")
      if not file_token:
          return {"error": "DAG has no file_token", "dag_id": dag_id}
  
      # Get source using file_token
      return self._call(f"dagSources/{file_token}")

• src/astro_airflow_mcp/adapters/base.py:220-222 (schema)

  Abstract base class definition for get_dag_source method that all adapters must implement. Defines the interface contract.

  @abstractmethod
  def get_dag_source(self, dag_id: str) -> dict[str, Any]:
      """Get source code of a DAG."""