Streamline D365 F&O Data Migration with DMF: Complete Guide to Package Export and Import

Moving data between Dynamics 365 Finance & Operations environments is a fundamental yet challenging task. Whether you're refreshing test data from production, deploying configuration changes, or replicating master data across legal entities, the Data Management Framework (DMF) is your go-to tool—but only if you use it correctly.

In this comprehensive guide, I'll walk you through the complete automation workflow for exporting and importing data packages using DMF, with a focus on proper entity sequencing, execution management, and error handling. This approach leverages MCP (Model Context Protocol) prompts to make complex DMF operations reproducible and reliable.

Visual Overview

DMF Workflow: 5-Phase Automation

Complete guide to automated data package migration

🔍

Entity Execution Sequence

Understand Dependencies

Automatically determine the correct import/export order based on entity relationships

GetEntitySequence OData action

Analyzes entity dependencies

Returns execution units, levels, and sequence

Prevents foreign key violations

End-to-End DMF Data Package Flow

From source export to target import with monitoring

👉 Scroll horizontally to view all steps

Action

Decision

Data

🏢

Source Environment

D365 F&O with data to export

🔢

Get Entity Sequence

GetEntitySequence action

📝

Create Export Project

DataManagementDefinitionGroups

📤

Execute Export

ExportToPackage action

📦

Data Package

ZIP file with entities

🔗

Get Download URL

GetExportedPackageUrl

🌐

Transfer Package

SAS URL (60-90 min)

📥

Import to Target

ImportFromPackageAsync

📊

Monitor Execution

GetExecutionSummaryStatus

✅

Target Environment

Data successfully imported

🏢

Source Environment

D365 F&O with data to export

🔢

Get Entity Sequence

GetEntitySequence action

📝

Create Export Project

DataManagementDefinitionGroups

📤

Execute Export

ExportToPackage action

📦

Data Package

ZIP file with entities

🔗

Get Download URL

GetExportedPackageUrl

🌐

Transfer Package

SAS URL (60-90 min)

📥

Import to Target

ImportFromPackageAsync

📊

Monitor Execution

GetExecutionSummaryStatus

✅

Target Environment

Data successfully imported

Phase 1

Sequence & Create

Steps 1-3

Phase 2

Export & Package

Steps 4-5

Phase 3

Retrieve URL

Steps 6-7

Phase 4

Import

Step 8

Phase 5

Monitor

Steps 9-10

The Challenge: Why DMF Data Migration Is Tricky

If you've ever tried to move data between D365 F&O environments, you've likely encountered these frustrations:

Dependency Hell: Import customers before customer groups? Wrong order = foreign key violations
Manual Sequence Configuration: Manually figuring out which entities depend on which is error-prone
Execution Tracking: "Is my export done yet?" followed by repeated UI refreshes
Error Investigation: Cryptic error messages buried deep in staging logs
URL Expiration: Package URLs expire in 60-90 minutes—miss the window, start over

These issues stem from DMF's flexibility: it's powerful but requires precise configuration. Get the entity sequence wrong, and your entire import fails. Miss a validation error, and you're troubleshooting in production.

The Solution: Automated DMF Workflow

The approach I'm sharing automates the entire DMF lifecycle using five specialized MCP prompts that handle:

Entity Execution Sequence Discovery - Automatically determine the correct import/export order
Export Package Creation - Build DMF projects with proper configuration
Export Execution & Package Retrieval - Trigger exports and get downloadable packages
Import to Target Environment - Load packages into any legal entity or environment
Monitoring & Error Handling - Track progress and handle failures systematically

Let's dive into each phase.

Phase 1: Understanding Entity Dependencies

Problem: DMF requires entities to be processed in dependency order. Import customer groups before customers, payment terms before invoices, and so on. Getting this wrong causes cascade failures.

Solution: The DMF Entity Execution Sequence Prompt automates dependency analysis.

How It Works

The GetEntitySequence OData action analyzes entity relationships and returns:

Execution Unit: Processing group number
Level in Execution Unit: Dependency level (lower = process first)
Sequence in Level: Order within the same level

Key Implementation Details

Critical Gotcha: Use entity labels (e.g., "Customers V3"), not entity names (e.g., "CustCustomerV3Entity") or collection names (e.g., "CustomersV3").

from d365fo_client import FOClient, FOClientConfig

# Initialize client
config = FOClientConfig(
    base_url="https://your-env.dynamics.com",
    use_default_credentials=True
)

async with FOClient(config) as client:
    # Step 1: Get entity labels from schema
    entity_info = await client.get_public_entity_info("CustomersV3")
    entity_label = entity_info.label_text  # "Customers V3"

    # Step 2: Call GetEntitySequence with labels
    sequence_response = await client.call_action(
        "Microsoft.Dynamics.DataEntities.GetEntitySequence",
        parameters={
            "listOfDataEntities": "Terms of payment,Customer groups,Customers V3"
        },
        entity_name="DataManagementDefinitionGroups"
    )

    # Response: "Terms of payment-1-1-1,Customer groups-1-2-1,Customers V3-1-3-1,"

Parsing the Sequence

The response is a comma-separated string with format:

EntityLabel-ExecutionUnit-LevelInExecutionUnit-SequenceInLevel

Parse this into structured metadata:

from dataclasses import dataclass
from typing import List

@dataclass
class EntitySequence:
    entity_label: str
    execution_unit: int
    level_in_execution_unit: int
    sequence_in_level: int

def parse_sequence(response: str) -> List[EntitySequence]:
    """Parse the sequence response into structured data"""
    entities = []

    for entry in response.split(','):
        if not entry.strip():
            continue

        parts = entry.split('-')
        if len(parts) == 4:
            entities.append(EntitySequence(
                entity_label=parts[0],
                execution_unit=int(parts[1]),
                level_in_execution_unit=int(parts[2]),
                sequence_in_level=int(parts[3])
            ))

    # Sort by execution order
    entities.sort(key=lambda x: (
        x.execution_unit,
        x.level_in_execution_unit,
        x.sequence_in_level
    ))

    return entities

Example Output

Execution Order:
1. Terms of payment (Unit: 1, Level: 1, Sequence: 1)
2. Terms of delivery (Unit: 1, Level: 1, Sequence: 1)
3. Customer groups (Unit: 1, Level: 2, Sequence: 1)
4. Customers V3 (Unit: 1, Level: 3, Sequence: 1)
5. Customer parameters (Unit: 1, Level: 4, Sequence: 1)

This sequence ensures payment terms and customer groups exist before importing customer records.

Learn More: DMF Entity Execution Sequence Prompt

Phase 2: Creating the Export Package

Problem: Manually creating DMF export projects through the UI is time-consuming and error-prone. You have to add entities one by one, configure formats, and hope you got the sequence right.

Solution: The DMF Create Export Package Prompt automates project creation with proper sequencing.

Workflow Overview

Get entity execution sequence (Phase 1)
Create DMF project definition
Add each entity with correct execution metadata
Verify project configuration

Step-by-Step Implementation

Step 1: Create Project Definition

Use the DataManagementDefinitionGroups entity:

from d365fo_client import FOClient, FOClientConfig

async with FOClient(config) as client:
    # Create project definition
    project_data = {
        "Name": "CustomerMasterExport_USMF_2025",
        "ProjectCategory": "Project",
        "OperationType": "Export",
        "GenerateDataPackage": "Yes",
        "Description": "Customer master data export from USMF",
        "TruncateEntityData": "Yes"
    }

    await client.create_entity(
        "DataManagementDefinitionGroups",
        project_data
    )

Field Explanations:

Name: Unique project identifier (acts as primary key)
OperationType: "Export" for exports, "Import" for imports
GenerateDataPackage: "Yes" creates downloadable ZIP package
TruncateEntityData: "Yes" clears previous export data

Step 2: Add Entities with Sequence

For each entity in the sequence, create a DataManagementDefinitionGroupDetails record:

# Assuming sequenced_entities is a list of EntitySequence objects
for entity in sequenced_entities:
    entity_detail = {
        "DefinitionGroupId": "CustomerMasterExport_USMF_2025",
        "EntityName": entity.entity_label,
        "ExecutionUnit": entity.execution_unit,
        "LevelInExecutionUnit": entity.level_in_execution_unit,
        "SequenceInLevel": entity.sequence_in_level,
        "SourceFormat": "EXCEL",
        "DefaultRefreshType": "FullPush",
        "AutoGenerateMapping": "Yes"
    }

    await client.create_entity(
        "DataManagementDefinitionGroupDetails",
        entity_detail
    )

Configuration Options:

SourceFormat: EXCEL, CSV, XML, PACKAGE, ODBC
DefaultRefreshType: "FullPush" (all records) or "Incremental" (changes only)
AutoGenerateMapping: "Yes" auto-maps fields, "No" for manual mapping

Step 3: Verify Project

Query the project to confirm all entities were added:

from d365fo_client import QueryOptions

# Get project record
project = await client.get_entity(
    "DataManagementDefinitionGroups",
    "CustomerMasterExport_USMF_2025"
)

# Get project entities
query_options = QueryOptions(
    filter="DefinitionGroupId eq 'CustomerMasterExport_USMF_2025'"
)

entities_result = await client.get_entities(
    "DataManagementDefinitionGroupDetails",
    query_options
)

entity_count = len(entities_result.get('value', []))
print(f"✅ Project created with {entity_count} entities")

Learn More: DMF Create Export Package Prompt

Phase 3: Executing the Export and Getting the Package

Problem: After creating the project, you need to trigger the export, wait for completion, and retrieve the package URL—all while dealing with asynchronous processing.

Solution: The DMF Execute Export Package Prompt handles execution and URL retrieval.

Two-Action Workflow

Action 1: ExportToPackage

Triggers the export job asynchronously:

import asyncio
from d365fo_client import FOClient, FOClientConfig

async with FOClient(config) as client:
    # Execute export
    export_params = {
        "definitionGroupId": "CustomerMasterExport_USMF_2025",
        "packageName": "CustomerMasterExport_USMF_2025 - USMF",
        "executionId": "CustomerMasterExport_USMF_2025-export-20251005-143022",
        "reExecute": True,
        "legalEntityId": "USMF"
    }

    export_result = await client.call_action(
        "Microsoft.Dynamics.DataEntities.ExportToPackage",
        parameters=export_params,
        entity_name="DataManagementDefinitionGroups"
    )

    # Extract execution ID from response
    if isinstance(export_result, dict) and "value" in export_result:
        execution_id = export_result["value"]
    else:
        execution_id = str(export_result)

    print(f"Export started: {execution_id}")

Parameter Notes:

packageName: Human-readable name (suggested: "your_project_name - your_company")
executionId: Unique identifier (suggested: "your_project_name-export-timestamp")
reExecute: true allows re-running, false prevents duplicates
legalEntityId: Company code to export from

Action 2: GetExportedPackageUrl

Retrieves the download URL after export completes:

import asyncio

# Poll for package URL
package_url = ""
retries = 0
max_retries = 20

while not package_url and retries < max_retries:
    await asyncio.sleep(5)  # Wait 5 seconds

    try:
        url_result = await client.call_action(
            "Microsoft.Dynamics.DataEntities.GetExportedPackageUrl",
            parameters={"executionId": execution_id},
            entity_name="DataManagementDefinitionGroups"
        )

        # Extract URL from response
        if isinstance(url_result, dict) and "value" in url_result:
            package_url = url_result["value"]
        elif isinstance(url_result, str):
            package_url = url_result

    except Exception as error:
        # Export still running
        pass

    retries += 1

if package_url:
    print(f"📦 Package ready: {package_url}")
    print(f"⏰ URL expires in ~90 minutes")
else:
    print(f"⚠️ Export taking longer than expected")

Timing Considerations

Export Duration:

Small datasets (< 1,000 records): Seconds
Medium datasets (1,000-50,000 records): 1-10 minutes
Large datasets (> 50,000 records): 10+ minutes

Polling Strategy:

Check immediately after export starts
If not ready, wait 5-10 seconds between retries
Maximum 20 retries (~2-3 minutes)
Inform user if taking longer

Package URL Characteristics

Temporary SAS URL with time-limited access (60-90 minutes)
Points to ZIP file in Azure Blob Storage
Contains Excel files (one per entity), manifest, and metadata
Download immediately to avoid expiration

Learn More: DMF Execute Export Package Prompt

Phase 4: Importing to Target Environment

Problem: Importing the package to a different legal entity or environment requires recreating the project configuration, mapping files, and executing the import—all manually.

Solution: The DMF Import Data Package Prompt automates the entire import process.

Single-Action Import

The beauty of DMF imports: one action does everything!

from d365fo_client import FOClient, FOClientConfig

async with FOClient(config) as client:
    import_params = {
        "packageUrl": "https://storage.blob.core.windows.net/.../DMFPackage.zip?sastoken",
        "definitionGroupId": "CustomerMasterImport_TDGP_2025",
        "executionId": "CustomerMasterImport_TDGP_2025-import-20251005-001",
        "execute": True,
        "overwrite": True,
        "legalEntityId": "TDGP",
        "failOnError": True,
        "runAsyncWithoutBatch": False,
        "thresholdToRunInBatch": 1000
    }

    await client.call_action(
        "Microsoft.Dynamics.DataEntities.ImportFromPackageAsync",
        parameters=import_params,
        entity_name="DataManagementDefinitionGroups"
    )

What Happens Behind the Scenes

Downloads Package: Retrieves ZIP from URL
Creates Import Project: Automatically creates new DMF project
Extracts Entities: Unpacks files and identifies entities
Stages Data: Loads data into staging tables
Validates Data: Runs business validation rules
Imports Data: Moves validated data to target tables
Preserves Sequence: Processes entities in dependency order

Parameter Deep Dive

Critical Parameters:

packageUrl: Full SAS URL from export (must not be expired!)
definitionGroupId: New project name (suggested: append "-Import" and target company)
executionId: Unique execution ID (suggested: "your_project_name-import-timestamp")
legalEntityId: Target company code (e.g., "TDGP", "USMF", "DAT")

Behavior Controls:

execute: true starts import immediately, false only creates project
overwrite: true updates existing records, false skips duplicates
failOnError: true stops on first error, false continues with warnings

Performance Controls:

runAsyncWithoutBatch: true bypasses batch framework (faster for > 1,000 records)
thresholdToRunInBatch: Record threshold for batch processing (default: 1,000)

Import Project Naming Best Practices

# Original export project
export_project = "CustomerMasterExport_2025"

# Import project naming strategies
import_project1 = f"{export_project}_Import_TDGP"  # Append company
import_project2 = f"{export_project}_Import_20251005"  # Append date
import_project3 = "CustomerMasterImport_TDGP_2025"  # Descriptive

Multi-Environment Import Pattern

Import the same package to multiple legal entities:

package_url = "https://storage.blob.core.windows.net/.../package.zip?sas"

# Import to TDGP
await import_from_package_async(package_url, "CustomerImport_TDGP", "TDGP")

# Import to DAT
await import_from_package_async(package_url, "CustomerImport_DAT", "DAT")

# Import to USMF
await import_from_package_async(package_url, "CustomerImport_USMF_Copy", "USMF")

async def import_from_package_async(package_url: str, project_id: str, legal_entity: str):
    """Helper function to import data package"""
    import_params = {
        "packageUrl": package_url,
        "definitionGroupId": project_id,
        "executionId": f"{project_id}-import-{int(time.time())}",
        "execute": True,
        "overwrite": True,
        "legalEntityId": legal_entity,
        "failOnError": True,
        "runAsyncWithoutBatch": False,
        "thresholdToRunInBatch": 1000
    }

    await client.call_action(
        "Microsoft.Dynamics.DataEntities.ImportFromPackageAsync",
        parameters=import_params,
        entity_name="DataManagementDefinitionGroups"
    )

Learn More: DMF Import Data Package Prompt

Phase 5: Monitoring and Error Handling

Problem: DMF operations run asynchronously. Without monitoring, you don't know if they succeeded, partially failed, or are still running. Errors are buried in staging logs with cryptic messages.

Solution: The DMF Monitoring and Error Handling Prompt provides comprehensive tracking and diagnostics.

The Four-Step Monitoring Workflow

Step 1: Check Overall Execution Status

from d365fo_client import FOClient, FOClientConfig

async with FOClient(config) as client:
    summary_result = await client.call_action(
        "Microsoft.Dynamics.DataEntities.GetExecutionSummaryStatus",
        parameters={"executionId": "CustomerMasterImport_TDGP_2025-import-20251005-001"},
        entity_name="DataManagementDefinitionGroups"
    )

    # Extract status value
    if isinstance(summary_result, dict) and "value" in summary_result:
        status = summary_result["value"]
    else:
        status = str(summary_result)

    print(f"Overall Status: {status}")

Status Values:

NotStarted: Execution queued but not yet running
InProgress: Currently processing entities
Succeeded: All entities imported/exported successfully
PartiallySucceeded: Some entities succeeded, some failed
Failed: Entire execution failed
Canceled: User canceled the execution

Step 2: Get Per-Entity Breakdown

When status is "PartiallySucceeded" or "Failed", drill down:

entity_list_result = await client.call_action(
    "Microsoft.Dynamics.DataEntities.GetEntityExecutionSummaryStatusList",
    parameters={"executionId": execution_id},
    entity_name="DataManagementDefinitionGroups"
)

# Extract entity list from response
if isinstance(entity_list_result, dict) and "value" in entity_list_result:
    entities = entity_list_result["value"]
else:
    entities = entity_list_result if isinstance(entity_list_result, list) else []

for entity in entities:
    total = entity.get("TotalRecords", 0)
    success = entity.get("SuccessRecords", 0)
    success_rate = (success / total * 100) if total > 0 else 0

    print(f"{entity.get('EntityName')}: {entity.get('Status')}")
    print(f"  {success_rate:.1f}% success ({entity.get('ErrorRecords', 0)} errors)")
    print(f"  Duration: {entity.get('ExecutionStartDateTime')} to {entity.get('ExecutionEndDateTime')}")

Example Output:

Terms of payment: PartiallySucceeded
  73.3% success (4 errors)
  Duration: 2025-10-05T20:10:00Z to 2025-10-05T20:10:15Z

Customer groups: Succeeded
  100.0% success (0 errors)
  Duration: 2025-10-05T20:10:15Z to 2025-10-05T20:10:18Z

Customers V3: PartiallySucceeded
  95.2% success (25 errors)
  Duration: 2025-10-05T20:10:18Z to 2025-10-05T20:11:45Z

Step 3: Get Detailed Error Messages

Extract specific error details for failed records:

import json

errors_result = await client.call_action(
    "Microsoft.Dynamics.DataEntities.GetExecutionErrors",
    parameters={"executionId": execution_id},
    entity_name="DataManagementDefinitionGroups"
)

# Extract and parse errors
if isinstance(errors_result, dict) and "value" in errors_result:
    errors_json = errors_result["value"]
else:
    errors_json = str(errors_result)

errors = json.loads(errors_json)

for index, error in enumerate(errors, 1):
    print(f"\nError {index}:")
    print(f"  Record: {error.get('RecordId')}")
    print(f"  Field: {error.get('Field') or 'N/A'}")
    print(f"  Message: {error.get('ErrorMessage')}")

Example Errors:

Error 1:
  Record: Cash
  Field:
  Message: Chart of accounts invalid: 0

Error 2:
  Record: Month+15
  Field:
  Message: Payment day '15th' not found

Error 3:
  Record: CUST-001
  Field: CustGroup
  Message: The value 'PREMIUM' in field 'CustGroup' is not found in the related table 'CustGroup'

Step 4: Generate Error Files

Create downloadable files with failed record details:

# Generate error keys file
await client.call_action(
    "Microsoft.Dynamics.DataEntities.GenerateImportTargetErrorKeysFile",
    parameters={"executionId": execution_id},
    entity_name="DataManagementDefinitionGroups"
)

# Get download URLs
error_keys_result = await client.call_action(
    "Microsoft.Dynamics.DataEntities.GetImportTargetErrorKeysFileUrl",
    parameters={"executionId": execution_id},
    entity_name="DataManagementDefinitionGroups"
)

staging_error_result = await client.call_action(
    "Microsoft.Dynamics.DataEntities.GetImportStagingErrorFileUrl",
    parameters={"executionId": execution_id},
    entity_name="DataManagementDefinitionGroups"
)

# Extract URLs
error_keys_url = error_keys_result.get("value") if isinstance(error_keys_result, dict) else str(error_keys_result)
staging_error_url = staging_error_result.get("value") if isinstance(staging_error_result, dict) else str(staging_error_result)

print(f"\nError Files:")
print(f"  Error Keys: {error_keys_url}")
print(f"  Staging Errors: {staging_error_url}")

Error Categorization Pattern

Categorize errors for better analysis:

from typing import Dict, List, Any

def categorize_errors(errors: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
    """Categorize errors by type for analysis"""
    categories = {
        "missing_reference": [],
        "validation": [],
        "business_rule": [],
        "system": [],
        "other": []
    }

    for error in errors:
        msg = error.get("ErrorMessage", "")

        if "is not found in the related table" in msg:
            categories["missing_reference"].append(error)
        elif "Validation" in msg or "failed" in msg:
            categories["validation"].append(error)
        elif "invalid" in msg or "passed into" in msg:
            categories["system"].append(error)
        else:
            categories["other"].append(error)

    print("\nError Analysis:")
    print(f"  Missing Reference Data: {len(categories['missing_reference'])}")
    print(f"  Validation Errors: {len(categories['validation'])}")
    print(f"  System/Config Errors: {len(categories['system'])}")
    print(f"  Other: {len(categories['other'])}")

    return categories

Common Error Resolution Patterns

Pattern 1: Missing Reference Data

Error: "The value 'X' in field 'Y' is not found in the related table 'Z'"
Resolution:
1. Export and import reference data first
2. Modify source data to use existing references
3. Create missing reference records manually

Pattern 2: Configuration Issues

Error: "Chart of accounts passed into findByMainAccountIdAndCOA() was invalid: 0"
Resolution:
1. Verify General Ledger setup
2. Ensure main accounts exist
3. Check posting profiles configuration

Pattern 3: Business Rule Violations

Error: "Validation of field 'X' failed"
Resolution:
1. Review validation rules for the entity
2. Correct data to meet validation criteria
3. Re-import corrected records

Learn More: DMF Monitoring and Error Handling Prompt

Complete End-to-End Example

Let's put it all together with a real-world scenario: exporting customer master data from USMF and importing to TDGP.

Scenario Setup

Source: USMF legal entity (Production)
Target: TDGP legal entity (Test)
Entities: Customer groups, Customers, Payment terms, Customer parameters

Step-by-Step Execution

import asyncio
import time
import json
from d365fo_client import FOClient, FOClientConfig, QueryOptions

async def run_complete_migration():
    """Complete end-to-end data migration example"""

    # Initialize client
    config = FOClientConfig(
        base_url="https://your-env.dynamics.com",
        use_default_credentials=True
    )

    async with FOClient(config) as client:
        # Phase 1: Get entity sequence
        entity_labels = [
            "Customer groups",
            "Customers V3",
            "Terms of payment",
            "Customer parameters"
        ]

        sequence_response = await client.call_action(
            "Microsoft.Dynamics.DataEntities.GetEntitySequence",
            parameters={"listOfDataEntities": ",".join(entity_labels)},
            entity_name="DataManagementDefinitionGroups"
        )

        # Parse sequence
        sequence_value = sequence_response.get("value") if isinstance(sequence_response, dict) else str(sequence_response)
        sequenced_entities = parse_sequence(sequence_value)
        print("✅ Entity sequence determined")

        # Phase 2: Create export project
        export_project = "CustomerMasterExport_USMF_2025"

        await client.create_entity(
            "DataManagementDefinitionGroups",
            {
                "Name": export_project,
                "ProjectCategory": "Project",
                "OperationType": "Export",
                "GenerateDataPackage": "Yes",
                "Description": "Customer master data export from USMF",
                "TruncateEntityData": "Yes"
            }
        )

        for entity in sequenced_entities:
            await client.create_entity(
                "DataManagementDefinitionGroupDetails",
                {
                    "DefinitionGroupId": export_project,
                    "EntityName": entity.entity_label,
                    "ExecutionUnit": entity.execution_unit,
                    "LevelInExecutionUnit": entity.level_in_execution_unit,
                    "SequenceInLevel": entity.sequence_in_level,
                    "SourceFormat": "EXCEL",
                    "DefaultRefreshType": "FullPush",
                    "AutoGenerateMapping": "Yes"
                }
            )

        print("✅ Export project created with 4 entities")

        # Phase 3: Execute export and get package
        export_execution_id = f"{export_project}-export-{int(time.time())}"

        await client.call_action(
            "Microsoft.Dynamics.DataEntities.ExportToPackage",
            parameters={
                "definitionGroupId": export_project,
                "packageName": f"{export_project} - USMF",
                "executionId": export_execution_id,
                "reExecute": True,
                "legalEntityId": "USMF"
            },
            entity_name="DataManagementDefinitionGroups"
        )

        print("✅ Export started")

        # Poll for package URL
        package_url = ""
        retries = 0

        while not package_url and retries < 20:
            await asyncio.sleep(5)
            try:
                url_result = await client.call_action(
                    "Microsoft.Dynamics.DataEntities.GetExportedPackageUrl",
                    parameters={"executionId": export_execution_id},
                    entity_name="DataManagementDefinitionGroups"
                )
                package_url = url_result.get("value") if isinstance(url_result, dict) else str(url_result)
            except:
                pass
            retries += 1

        print(f"✅ Package ready: {package_url}")

        # Phase 4: Import to target legal entity
        import_project = "CustomerMasterImport_TDGP_2025"
        import_execution_id = f"{import_project}-import-{int(time.time())}"

        await client.call_action(
            "Microsoft.Dynamics.DataEntities.ImportFromPackageAsync",
            parameters={
                "packageUrl": package_url,
                "definitionGroupId": import_project,
                "executionId": import_execution_id,
                "execute": True,
                "overwrite": True,
                "legalEntityId": "TDGP",
                "failOnError": True,
                "runAsyncWithoutBatch": False,
                "thresholdToRunInBatch": 1000
            },
            entity_name="DataManagementDefinitionGroups"
        )

        print("✅ Import started")

        # Phase 5: Monitor import
        await asyncio.sleep(10)  # Wait for import to start

        summary_result = await client.call_action(
            "Microsoft.Dynamics.DataEntities.GetExecutionSummaryStatus",
            parameters={"executionId": import_execution_id},
            entity_name="DataManagementDefinitionGroups"
        )

        status = summary_result.get("value") if isinstance(summary_result, dict) else str(summary_result)

        if status == "Succeeded":
            print("✅ Import completed successfully")
        elif status == "PartiallySucceeded":
            print("⚠️ Import partially succeeded, checking errors...")

            entity_list_result = await client.call_action(
                "Microsoft.Dynamics.DataEntities.GetEntityExecutionSummaryStatusList",
                parameters={"executionId": import_execution_id},
                entity_name="DataManagementDefinitionGroups"
            )

            entities = entity_list_result.get("value", []) if isinstance(entity_list_result, dict) else []

            for entity in entities:
                if entity.get("ErrorRecords", 0) > 0:
                    print(f"  ❌ {entity.get('EntityName')}: {entity.get('ErrorRecords')} errors")
                else:
                    print(f"  ✅ {entity.get('EntityName')}: Success")

            # Get detailed errors
            errors_result = await client.call_action(
                "Microsoft.Dynamics.DataEntities.GetExecutionErrors",
                parameters={"executionId": import_execution_id},
                entity_name="DataManagementDefinitionGroups"
            )

            errors_json = errors_result.get("value") if isinstance(errors_result, dict) else str(errors_result)
            errors = json.loads(errors_json)

            print(f"\nDetailed Errors: {len(errors)} total")
        elif status == "InProgress":
            print("⏳ Import still running, check status later")
        else:
            print(f"❌ Import failed with status: {status}")

# Run the migration
if __name__ == "__main__":
    asyncio.run(run_complete_migration())

Expected Output

✅ Entity sequence determined
✅ Export project created with 4 entities
✅ Export started
✅ Package ready: https://storage.blob.core.windows.net/.../DMFPackage.zip?sv=...
✅ Import started
✅ Import completed successfully

Use Cases and Best Practices

Use Case 1: Environment Data Refresh

Scenario: Refresh test environment with production data monthly

Implementation:

import asyncio
from d365fo_client import FOClient, FOClientConfig

async def monthly_refresh():
    """Schedule monthly data refresh from production to test"""
    source_env = "PROD-USMF"
    target_env = "TEST-USMF"

    entities = ["Customers V3", "Vendors V2", "Released products V2"]

    async with FOClient(config) as client:
        # Export from production
        package_url = await export_data_package(
            client, "MonthlyRefresh", source_env, entities
        )

        # Import to test
        await import_data_package(
            client, package_url, "MonthlyRefresh_Test", target_env
        )

async def export_data_package(client, project_name, legal_entity, entities):
    """Helper function to export data package"""
    # Get entity sequence
    sequence_response = await client.call_action(
        "Microsoft.Dynamics.DataEntities.GetEntitySequence",
        parameters={"listOfDataEntities": ",".join(entities)},
        entity_name="DataManagementDefinitionGroups"
    )

    # Create export project and entities
    # ... (implementation from Phase 2)

    # Execute export
    # ... (implementation from Phase 3)

    return package_url

async def import_data_package(client, package_url, project_name, legal_entity):
    """Helper function to import data package"""
    # ... (implementation from Phase 4)
    pass

Best Practices:

Schedule during low-usage hours
Use overwrite: true for full refresh
Monitor for errors and validate critical records
Archive package URLs for audit trail

Use Case 2: Cross-Company Configuration Deployment

Scenario: Deploy standard configuration to multiple legal entities

Implementation:

async def deploy_config_to_companies():
    """Deploy standard configuration to multiple companies"""
    config_package_url = await export_data_package(
        client,
        "StandardConfig",
        "CONFIG",
        ["Customer parameters", "Accounts receivable parameters", "Number sequences"]
    )

    target_companies = ["TDGP", "DAT", "WEST"]

    for company in target_companies:
        await import_data_package(
            client,
            config_package_url,
            f"StandardConfig_{company}",
            company
        )

Best Practices:

Export from "golden" configuration company
Use descriptive project names per target
Set failOnError: true for critical configuration
Verify each import before moving to next company

Use Case 3: Selective Data Migration

Scenario: Migrate specific customer segment to new legal entity

Implementation:

async def migrate_customer_segment():
    """Export with filter (requires custom view or staging)"""
    async with FOClient(config) as client:
        # Note: Filtering requires custom DMF project setup
        # with filtered views or staging queries
        segment_package_url = await export_data_package(
            client,
            "PremiumCustomers",
            "USMF",
            ["Customer groups", "Customers V3"]
        )

        # Import to dedicated legal entity
        await import_data_package(
            client,
            segment_package_url,
            "PremiumCustomers_PREM",
            "PREM"
        )

Best Practices:

Include all dependent reference data
Use overwrite: false for incremental loads
Test with small subset first
Document filtering criteria

Use Case 4: Data Backup and Disaster Recovery

Scenario: Regular backups for compliance and disaster recovery

Implementation:

async def backup_critical_data():
    """Regular backups for compliance and disaster recovery"""
    from datetime import date

    backup_date = date.today().isoformat()

    async with FOClient(config) as client:
        backup_package_url = await export_data_package(
            client,
            f"Backup_{backup_date}",
            "PROD",
            get_all_critical_entities()
        )

        # Store URL in secure backup system
        await archive_package_url({
            "url": backup_package_url,
            "timestamp": backup_date,
            "legal_entity": "PROD",
            "retention_days": 90
        })

def get_all_critical_entities():
    """Get list of all critical entities for backup"""
    return [
        "Customers V3",
        "Vendors V2",
        "Released products V2",
        # ... add all critical entities
    ]

async def archive_package_url(backup_info: dict):
    """Store package URL in secure backup system"""
    # Implementation for your backup storage
    pass

Best Practices:

Export all transactional and master data
Store packages in secure, geo-redundant storage
Test restore process quarterly
Maintain 90-day backup history minimum

Common Pitfalls and How to Avoid Them

Pitfall 1: Wrong Entity Sequence

Problem: Importing customers before customer groups causes foreign key errors.

Solution: Always use GetEntitySequence to determine order. Never guess!

Pitfall 2: Expired Package URLs

Problem: SAS URLs expire in 60-90 minutes, causing import failures.

Solution:

Download packages immediately after export
Store in your own blob storage for long-term use
Re-export if URL expires before import

Pitfall 3: Ignoring "PartiallySucceeded" Status

Problem: Assuming "PartiallySucceeded" means "good enough" leads to incomplete data.

Solution:

Always check entity-level status
Investigate and resolve all errors
Re-import corrected records
Only accept "Succeeded" for production imports

Pitfall 4: Missing Reference Data

Problem: Importing transactions before master data fails validation.

Solution:

Export and import in logical groups:
1. System configuration
2. Reference data (groups, terms, categories)
3. Master data (customers, vendors, products)
4. Transactions (orders, invoices)

Pitfall 5: Not Monitoring Long-Running Imports

Problem: Large imports take hours; without monitoring, failures go unnoticed.

Solution:

Implement polling with GetExecutionSummaryStatus
Set up notifications for completion/failure
Use GetMessageStatus to check queue progress
Log execution IDs for troubleshooting

Performance Optimization Tips

Tip 1: Batch vs. Non-Batch Processing

Small Datasets (fewer than 1,000 records):

import_params = {
    "runAsyncWithoutBatch": True,  # Faster startup
    "thresholdToRunInBatch": 5000
    # ... other parameters
}

Large Datasets (more than 50,000 records):

import_params = {
    "runAsyncWithoutBatch": False,  # Use batch framework
    "thresholdToRunInBatch": 1000   # Lower threshold
    # ... other parameters
}

Tip 2: Split Large Projects

Instead of one massive project with 50 entities:

# Split into logical groups
await export_and_import("MasterData", [
    "Customer groups",
    "Customers V3",
    "Payment terms"
])

await export_and_import("Transactions", [
    "Sales orders",
    "Sales invoices"
])

Benefits:

Faster execution per project
Easier error isolation
Parallel processing possible

Tip 3: Use Incremental Refresh

For ongoing synchronization:

# Configure entity for incremental refresh
entity_details = {
    "DefaultRefreshType": "Incremental",  # Only changed records
    # ... other fields
    # Requires change tracking enabled on entity
}

await client.create(
    "DataManagementDefinitionGroupDetails",
    entity_details
)

Requirements:

Entity must support change tracking
Baseline full import must complete first
Best for ongoing sync, not initial load

Security and Compliance Considerations

SAS URL Security

Issue: Package URLs contain access tokens that grant temporary access to data.

Best Practices:

Treat URLs as sensitive credentials
Never log URLs in plaintext
Use secure parameter storage (Azure Key Vault)
Monitor URL access in audit logs

import re

def mask_url(url: str) -> str:
    """Mask SAS token in URL for logging"""
    # Use regex to replace sig parameter
    masked = re.sub('sig=[^&]+', 'sig=***MASKED***', url)
    return masked

# Bad - don't log full URLs
print(f"Package URL: {package_url}")

# Good - mask sensitive tokens
print(f"Package URL: {mask_url(package_url)}")
logger.info("Package generated", extra={
    "execution_id": execution_id,
    "expires_in": "90m"
})

Audit Trail

Track all DMF operations:

from datetime import datetime
import logging

async def audited_import(client: FOClient, params: dict):
    """Import with audit trail"""
    audit_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "action": "DMF_IMPORT",
        "user": get_current_user(),
        "source_project": params["definition_group_id"],
        "target_legal_entity": params["legal_entity_id"],
        "execution_id": params["execution_id"]
    }

    # Log audit entry
    logging.info("DMF Import Started", extra=audit_entry)

    try:
        result = await import_data_package(client, **params)
        audit_entry["status"] = "SUCCESS"
        audit_entry["result"] = result
        return result
    except Exception as e:
        audit_entry["status"] = "FAILURE"
        audit_entry["error"] = str(e)
        raise
    finally:
        # Store in audit database
        await store_audit_entry(audit_entry)

Data Privacy

GDPR/Privacy Considerations:

Export only necessary fields (use field mappings)
Mask sensitive data before export
Encrypt packages at rest and in transit
Implement retention policies for packages
Document data flows for compliance

# Define sensitive fields to mask
sensitive_fields = [
    "SocialSecurityNumber",
    "BankAccountNumber",
    "CreditCardNumber"
]

async def mask_sensitive_data_in_staging(fields: list[str]):
    """Pre-export processing to mask sensitive data"""
    async with FOClient(config) as client:
        for field in fields:
            # Update staging table with masked values
            await client.update_staging_data(field, "***MASKED***")

Integration with CI/CD Pipelines

Azure DevOps Pipeline Example

trigger:
  branches:
    include:
      - main

stages:
  - stage: Export
    jobs:
      - job: ExportFromProd
        steps:
          - task: NodeScript@1
            inputs:
              scriptType: 'inline'
              inlineScript: |
                const packageUrl = await exportDataPackage(
                  "$(Build.BuildId)",
                  "PROD-USMF",
                  $(entityList)
                );
                console.log(`##vso[task.setvariable variable=packageUrl;isOutput=true]${packageUrl}`);

  - stage: Import
    dependsOn: Export
    jobs:
      - job: ImportToTest
        variables:
          packageUrl: $[ stageDependencies.Export.ExportFromProd.outputs['packageUrl'] ]
        steps:
          - task: NodeScript@1
            inputs:
              scriptType: 'inline'
              inlineScript: |
                await importDataPackage(
                  "$(packageUrl)",
                  "CI_Import_$(Build.BuildId)",
                  "TEST-USMF"
                );

                // Monitor and fail pipeline if errors
                const status = await monitorImport(executionId);
                if (status !== "Succeeded") {
                  throw new Error("Import failed or partially succeeded");
                }

Troubleshooting Guide

Issue: GetEntitySequence Returns Empty

Symptoms: Empty or malformed sequence response

Causes:

Using entity names instead of labels
Invalid entity names
Entity not available in DMF

Resolution:

# Wrong
params = {"listOfDataEntities": "CustCustomerV3Entity,CustGroup"}

# Correct
params = {"listOfDataEntities": "Customers V3,Customer groups"}

# Verify entity exists and get label
async with FOClient(config) as client:
    schema = await client.get_public_entity_info("CustomersV3")
    print(f"Entity label: {schema.label_text}")

Issue: Import Fails with "Package URL Not Accessible"

Symptoms: Import fails immediately with URL error

Causes:

SAS URL expired
Network connectivity issue
Invalid URL format

Resolution:

from urllib.parse import urlparse, parse_qs
from datetime import datetime
import httpx

async def check_package_url(package_url: str) -> bool:
    """Check if package URL is valid and accessible"""
    # Check URL expiration
    parsed = urlparse(package_url)
    params = parse_qs(parsed.query)

    if 'se' in params:
        expiry = params['se'][0]
        expiry_date = datetime.fromisoformat(expiry.replace('Z', '+00:00'))
        print(f"URL expires: {expiry}")

        if expiry_date < datetime.now(expiry_date.tzinfo):
            print("URL expired, need to re-export")
            return False

    # Test URL accessibility
    async with httpx.AsyncClient() as http_client:
        response = await http_client.head(package_url)
        if response.status_code != 200:
            raise Exception(f"URL not accessible: {response.status_code}")

    return True

# Re-export if expired
if not await check_package_url(package_url):
    async with FOClient(config) as client:
        package_url = await re_export_package(client, export_project)

Issue: Import Partially Succeeds with Validation Errors

Symptoms: "PartiallySucceeded" status with validation errors

Causes:

Missing reference data
Business rule violations
Configuration issues

Resolution:

import json

async def analyze_and_fix_errors(client: FOClient, execution_id: str):
    """Analyze errors by category and suggest fixes"""
    # Get execution errors
    errors_response = await client.call_action(
        "Microsoft.Dynamics.DataEntities.GetExecutionErrors",
        parameters={"executionId": execution_id},
        entity_name="DataManagementDefinitionGroups"
    )

    errors = json.loads(errors_response["value"])
    categorized = categorize_errors(errors)

    # Handle missing reference data
    if categorized["missing_reference"]:
        print("Import reference data first:")
        for err in categorized["missing_reference"]:
            print(f"  - {err['ErrorMessage']}")

        # Extract missing reference entities
        missing_entities = extract_missing_entities(categorized["missing_reference"])
        await export_and_import_references(client, missing_entities)

        # Retry original import
        await retry_import(client, execution_id)

def categorize_errors(errors: list) -> dict:
    """Categorize errors by type"""
    categories = {
        "missing_reference": [],
        "validation": [],
        "other": []
    }

    for error in errors:
        msg = error.get("ErrorMessage", "").lower()
        if "foreign key" in msg or "reference" in msg:
            categories["missing_reference"].append(error)
        elif "validation" in msg or "invalid" in msg:
            categories["validation"].append(error)
        else:
            categories["other"].append(error)

    return categories

Issue: Export Takes Extremely Long

Symptoms: Export runs for hours without completing

Causes:

Very large dataset
Complex entity with calculated fields
System load

Resolution:

async def check_export_progress(client: FOClient, export_execution_id: str):
    """Check message queue status to diagnose slow exports"""
    message_status = await client.call_action(
        "Microsoft.Dynamics.DataEntities.GetMessageStatus",
        parameters={"executionId": export_execution_id},
        entity_name="DataManagementDefinitionGroups"
    )

    print(f"Queued: {message_status['QueuedMessages']}")
    print(f"Processing: {message_status['ProcessingMessages']}")
    print(f"Processed: {message_status['ProcessedMessages']}")
    print(f"Batch Status: {message_status['BatchJobStatus']}")

    # If stuck, consider:
    # 1. Split into smaller projects
    # 2. Filter data at entity level
    # 3. Use batch framework (runAsyncWithoutBatch: false)
    # 4. Check system resources and batch server capacity

Conclusion

Automating D365 Finance & Operations data migration with DMF doesn't have to be painful. By following this structured approach—proper entity sequencing, automated project creation, execution management, and comprehensive error handling—you can make data package export and import operations reliable and repeatable.

Key Takeaways

Always use GetEntitySequence to determine correct entity order—never guess dependencies
Automate project creation with proper configuration to eliminate manual errors
Monitor executions systematically using status checks, entity breakdowns, and error analysis
Handle partial successes proactively by investigating errors and re-importing corrections
Secure package URLs and treat them as sensitive credentials with expiration awareness

Next Steps

Ready to implement this workflow? Check out the comprehensive MCP prompts:

Before diving into DMF automation, you might want to check out these related posts about D365 F&O automation and integration:

Building the Future of D365 F&O Integration - Learn about the d365fo-client Python library and MCP server that powers the automation examples in this post
D365FO MCP Server in Action - Watch a practical demo of setting up and using the MCP server for customer intelligence queries
One-Click Azure Deployment - Deploy D365FO MCP Server to Azure Container Apps with automated scripts and API key authentication
Enterprise-Grade Credential Management - Deep dive into advanced authentication, Azure Key Vault integration, and sync session architecture

Get in Touch

Need help implementing DMF automation in your D365 F&O environment? Want to discuss custom integration solutions or enterprise AI consulting?

Connect with me:

📧 Email: [email protected]
🐦 Twitter/X: @TheDataGuyPro
💼 LinkedIn: Muhammad Afzaal
💻 GitHub: @mafzaal
🎥 YouTube: @TheDataGuyPro
🎧 Podcast: TheDataGuy Show

Whether you're looking for consulting services, training, or just want to discuss D365 F&O automation strategies, I'd love to hear from you!

Visual Overview

DMF Workflow: 5-Phase Automation

Entity Execution Sequence

End-to-End DMF Data Package Flow

Contents

The Challenge: Why DMF Data Migration Is Tricky

The Solution: Automated DMF Workflow

Phase 1: Understanding Entity Dependencies

How It Works

Key Implementation Details

Parsing the Sequence

Example Output

Phase 2: Creating the Export Package

Workflow Overview

Step-by-Step Implementation

Step 1: Create Project Definition

Step 2: Add Entities with Sequence

Step 3: Verify Project

Phase 3: Executing the Export and Getting the Package

Two-Action Workflow

Action 1: ExportToPackage

Action 2: GetExportedPackageUrl

Timing Considerations

Package URL Characteristics

Phase 4: Importing to Target Environment

Single-Action Import

What Happens Behind the Scenes

Parameter Deep Dive

Import Project Naming Best Practices

Multi-Environment Import Pattern

Phase 5: Monitoring and Error Handling

The Four-Step Monitoring Workflow

Step 1: Check Overall Execution Status

Step 2: Get Per-Entity Breakdown

Step 3: Get Detailed Error Messages

Step 4: Generate Error Files

Error Categorization Pattern

Common Error Resolution Patterns

Complete End-to-End Example

Scenario Setup

Step-by-Step Execution

Expected Output

Use Cases and Best Practices

Use Case 1: Environment Data Refresh

Use Case 2: Cross-Company Configuration Deployment

Use Case 3: Selective Data Migration

Use Case 4: Data Backup and Disaster Recovery

Common Pitfalls and How to Avoid Them

Pitfall 1: Wrong Entity Sequence

Pitfall 2: Expired Package URLs

Pitfall 3: Ignoring "PartiallySucceeded" Status

Pitfall 4: Missing Reference Data

Pitfall 5: Not Monitoring Long-Running Imports

Performance Optimization Tips

Tip 1: Batch vs. Non-Batch Processing

Tip 2: Split Large Projects

Tip 3: Use Incremental Refresh

Security and Compliance Considerations

SAS URL Security

Audit Trail

Data Privacy

Integration with CI/CD Pipelines

Azure DevOps Pipeline Example

Troubleshooting Guide

Issue: GetEntitySequence Returns Empty

Issue: Import Fails with "Package URL Not Accessible"

Issue: Import Partially Succeeds with Validation Errors

Issue: Export Takes Extremely Long

Conclusion

Key Takeaways

Next Steps

Related Posts

Get in Touch