Dynamics 365

<!-- AUTO-GENERATED by export-skills.py — DO NOT EDIT -->
---
name: dynamics-365
description: Microsoft Dynamics 365 integration patterns for Sales, Service, Finance, and Supply Chain. Use when building AI agents that interact with Dataverse Web API (OData), F&O data entities, or integrate Azure OpenAI with Dynamics data.
tags: [dynamics-365, dataverse, microsoft, crm, erp]
---

# Microsoft Dynamics 365 Integration

Connect AI agents to Dynamics 365 (Sales, Service, Finance, Supply Chain) using Dataverse Web API (OData v4), F&O OData, and Azure OpenAI integration.

## When to Use

- Building AI agents that query or manage CRM data (leads, opportunities, cases, accounts)
- Integrating with Dynamics 365 Finance & Operations (journals, invoices, purchase orders)
- Combining Azure OpenAI with Dynamics data for deal analysis, case classification, or forecasting
- Using FetchXML for complex queries beyond OData $filter capability
- Triggering Power Automate flows from external AI agents

## API Surface

| Product | API Base URL | Style |
|---|---|---|
| **Sales / Service / Marketing** | `https://{org}.crm.dynamics.com/api/data/v9.2/` | Dataverse OData v4 |
| **Finance & Operations** | `https://{org}.cloudax.dynamics.com/data/` | F&O OData v4 |
| **Business Central** | `https://api.businesscentral.dynamics.com/v2.0/{tenant}/{env}/api/v2.0/` | BC OData v4 |

## Key Entities

### Sales

| Entity | OData Set | Key Fields |
|---|---|---|
| Account | `accounts` | `name`, `revenue`, `industrycode` |
| Contact | `contacts` | `fullname`, `emailaddress1`, `telephone1` |
| Lead | `leads` | `fullname`, `companyname`, `estimatedvalue`, `statuscode` |
| Opportunity | `opportunities` | `name`, `estimatedvalue`, `closeprobability`, `estimatedclosedate` |
| Quote | `quotes` | `name`, `totalamount` |
| Invoice | `invoices` | `name`, `totalamount`, `statuscode` |

### Customer Service

| Entity | OData Set | Key Fields |
|---|---|---|
| Case | `incidents` | `title`, `ticketnumber`, `prioritycode`, `statuscode` |
| Knowledge Article | `knowledgearticles` | `title`, `content`, `statecode` |
| Queue | `queues` | `name` |

### Finance & Operations

| Entity | OData Set | Key Fields |
|---|---|---|
| Journal Headers | `GeneralJournalHeaders` | Journal batch headers |
| Journal Lines | `GeneralJournalAccountEntries` | Line entries |
| Vendor Invoices | `VendorInvoiceHeaders` | AP invoices |
| Purchase Orders | `PurchaseOrderHeaders` | Procurement |
| Customers | `CustomersV3` | AR customers |
| Vendors | `VendorsV2` | AP vendors |
| Chart of Accounts | `MainAccounts` | GL accounts |

## Patterns

### 1. Authentication (Entra ID / OAuth 2.0)

```python
import msal
import requests

class Dynamics365Client:
    """Dynamics 365 client using MSAL (Entra ID) client credentials."""

    def __init__(self, tenant_id: str, client_id: str, client_secret: str, crm_url: str):
        self.crm_url = crm_url.rstrip("/")
        self.api_url = f"{self.crm_url}/api/data/v9.2"
        self._app = msal.ConfidentialClientApplication(
            client_id,
            authority=f"https://login.microsoftonline.com/{tenant_id}",
            client_credential=client_secret,
        )
        self._scope = [f"{self.crm_url}/.default"]
        self._session = requests.Session()
        self._session.headers.update({
            "OData-MaxVersion": "4.0",
            "OData-Version": "4.0",
            "Accept": "application/json",
            "Content-Type": "application/json; charset=utf-8",
            "Prefer": "odata.include-annotations=*",
        })

    def _ensure_token(self):
        token = self._app.acquire_token_for_client(scopes=self._scope)
        if "access_token" not in token:
            raise Exception(f"Auth failed: {token.get('error_description')}")
        self._session.headers["Authorization"] = f"Bearer {token['access_token']}"

    def get(self, entity_set: str, params: dict = None) -> dict:
        self._ensure_token()
        resp = self._session.get(f"{self.api_url}/{entity_set}", params=params)
        resp.raise_for_status()
        return resp.json()

    def get_by_id(self, entity_set: str, record_id: str, select: str = None) -> dict:
        self._ensure_token()
        params = {"$select": select} if select else None
        resp = self._session.get(f"{self.api_url}/{entity_set}({record_id})", params=params)
        resp.raise_for_status()
        return resp.json()

    def create(self, entity_set: str, data: dict) -> str:
        self._ensure_token()
        resp = self._session.post(f"{self.api_url}/{entity_set}", json=data)
        resp.raise_for_status()
        return resp.headers.get("OData-EntityId", "")

    def update(self, entity_set: str, record_id: str, data: dict) -> None:
        self._ensure_token()
        resp = self._session.patch(f"{self.api_url}/{entity_set}({record_id})", json=data)
        resp.raise_for_status()

    def delete(self, entity_set: str, record_id: str) -> None:
        self._ensure_token()
        resp = self._session.delete(f"{self.api_url}/{entity_set}({record_id})")
        resp.raise_for_status()
```

### 2. Sales -- Leads, Opportunities, Accounts

```python
# Query high-value opportunities with account expansion
def get_top_opportunities(client: Dynamics365Client, min_value: float = 10000) -> list[dict]:
    return client.get("opportunities", params={
        "$select": "name,estimatedvalue,closeprobability,estimatedclosedate",
        "$filter": f"statecode eq 0 and estimatedvalue gt {min_value}",
        "$orderby": "estimatedvalue desc",
        "$top": 20,
        "$expand": "parentaccountid($select=name,revenue)",
    }).get("value", [])

# Create a new lead
def create_lead(client: Dynamics365Client, first_name: str, last_name: str,
                company: str, email: str, estimated_value: float = 0) -> str:
    return client.create("leads", {
        "firstname": first_name,
        "lastname": last_name,
        "companyname": company,
        "emailaddress1": email,
        "estimatedvalue": estimated_value,
        "subject": f"Interest from {company}",
    })

# Qualify a lead (bound action)
def qualify_lead(client: Dynamics365Client, lead_id: str) -> dict:
    client._ensure_token()
    resp = client._session.post(
        f"{client.api_url}/leads({lead_id})/Microsoft.Dynamics.CRM.QualifyLead",
        json={"CreateAccount": True, "CreateContact": True, "CreateOpportunity": True, "Status": 3},
    )
    resp.raise_for_status()
    return resp.json()

# Update opportunity stage
def update_opportunity_stage(client: Dynamics365Client, opp_id: str,
                              stage: str, probability: int) -> None:
    client.update("opportunities", opp_id, {
        "stepname": stage,
        "closeprobability": probability,
    })
```

### 3. Customer Service -- Cases and Knowledge

```python
# Get open cases sorted by priority
def get_open_cases(client: Dynamics365Client, limit: int = 25) -> list[dict]:
    return client.get("incidents", params={
        "$select": "title,ticketnumber,prioritycode,statuscode,createdon",
        "$filter": "statecode eq 0",
        "$orderby": "prioritycode asc,createdon desc",
        "$top": limit,
        "$expand": "customerid_contact($select=fullname,emailaddress1)",
    }).get("value", [])

# Create a support case
def create_case(client: Dynamics365Client, title: str, description: str,
                customer_id: str, priority: int = 2) -> str:
    return client.create("incidents", {
        "title": title,
        "description": description,
        "prioritycode": priority,
        "customerid_contact@odata.bind": f"/contacts({customer_id})",
    })

# Search knowledge articles
def search_knowledge(client: Dynamics365Client, query: str) -> list[dict]:
    return client.get("knowledgearticles", params={
        "$select": "title,description,articlepublicnumber",
        "$filter": f"statecode eq 3 and contains(title,'{query}')",
        "$top": 10,
    }).get("value", [])
```

### 4. Finance & Operations

```python
class D365FinOpsClient:
    """Dynamics 365 Finance & Operations OData client."""

    def __init__(self, tenant_id: str, client_id: str, client_secret: str, fo_url: str):
        self.fo_url = fo_url.rstrip("/")
        self.data_url = f"{self.fo_url}/data"
        self._app = msal.ConfidentialClientApplication(
            client_id,
            authority=f"https://login.microsoftonline.com/{tenant_id}",
            client_credential=client_secret,
        )
        self._scope = [f"{self.fo_url}/.default"]
        self._session = requests.Session()
        self._session.headers.update({
            "Content-Type": "application/json",
            "Accept": "application/json",
        })

    def _ensure_token(self):
        token = self._app.acquire_token_for_client(scopes=self._scope)
        self._session.headers["Authorization"] = f"Bearer {token['access_token']}"

    def get(self, entity: str, params: dict = None) -> list[dict]:
        self._ensure_token()
        resp = self._session.get(f"{self.data_url}/{entity}", params=params)
        resp.raise_for_status()
        return resp.json().get("value", [])

# Query vendor invoices
def get_vendor_invoices(client: D365FinOpsClient, from_date: str) -> list[dict]:
    return client.get("VendorInvoiceHeaders", params={
        "$filter": f"InvoiceDate gt {from_date}",
        "$select": "InvoiceId,VendorName,InvoiceAmount,InvoiceDate",
        "$top": 50,
        "$orderby": "InvoiceDate desc",
    })

# Query journal entries
def get_journal_entries(client: D365FinOpsClient, from_date: str) -> list[dict]:
    return client.get("GeneralJournalAccountEntries", params={
        "$filter": f"PostingDate gt {from_date}",
        "$select": "JournalNumber,MainAccount,TransactionCurrencyAmount",
        "$top": 100,
    })

# Create a purchase order (F&O requires dataAreaId)
def create_purchase_order(client: D365FinOpsClient, po_number: str,
                          vendor: str, legal_entity: str = "usmf") -> dict:
    client._ensure_token()
    resp = client._session.post(f"{client.data_url}/PurchaseOrderHeaders", json={
        "PurchaseOrderNumber": po_number,
        "OrderVendorAccountNumber": vendor,
        "dataAreaId": legal_entity,
    })
    resp.raise_for_status()
    return resp.json()
```

### 5. FetchXML (Complex Queries)

```python
import urllib.parse

def fetch_xml_query(client: Dynamics365Client, entity_set: str, fetch_xml: str) -> list[dict]:
    """Execute a FetchXML query for complex joins and aggregations."""
    client._ensure_token()
    encoded = urllib.parse.quote(fetch_xml)
    resp = client._session.get(f"{client.api_url}/{entity_set}?fetchXml={encoded}")
    resp.raise_for_status()
    return resp.json().get("value", [])

# High-value opportunities in a specific industry
opps = fetch_xml_query(client, "opportunities", """
<fetch top="50">
  <entity name="opportunity">
    <attribute name="name"/>
    <attribute name="estimatedvalue"/>
    <attribute name="closeprobability"/>
    <filter>
      <condition attribute="statecode" operator="eq" value="0"/>
      <condition attribute="estimatedvalue" operator="gt" value="50000"/>
    </filter>
    <link-entity name="account" from="accountid" to="parentaccountid" alias="acct">
      <attribute name="name"/>
      <filter>
        <condition attribute="industrycode" operator="eq" value="6"/>
      </filter>
    </link-entity>
    <order attribute="estimatedvalue" descending="true"/>
  </entity>
</fetch>
""")
```

### 6. Change Tracking (Delta Sync)

```python
def get_changes(client: Dynamics365Client, entity_set: str,
                delta_link: str = None) -> tuple[list[dict], str]:
    """Track record changes since last sync.

    Returns (changed_records, next_delta_link).
    """
    client._ensure_token()
    if delta_link:
        resp = client._session.get(delta_link)
    else:
        resp = client._session.get(
            f"{client.api_url}/{entity_set}",
            headers={"Prefer": "odata.track-changes"},
        )
    resp.raise_for_status()
    data = resp.json()
    return data.get("value", []), data.get("@odata.deltaLink", "")

# Initial sync
records, delta = get_changes(client, "accounts")

# Subsequent sync -- only get changes
changed, delta = get_changes(client, "accounts", delta_link=delta)
```

### 7. Pagination Helper

```python
def get_all_records(client: Dynamics365Client, entity_set: str,
                    params: dict = None) -> list[dict]:
    """Handle server-side paging (5000 record default page size)."""
    client._ensure_token()
    records = []
    url = f"{client.api_url}/{entity_set}"
    while url:
        resp = client._session.get(url, params=params)
        resp.raise_for_status()
        data = resp.json()
        records.extend(data.get("value", []))
        url = data.get("@odata.nextLink")
        params = None  # nextLink already contains params
    return records
```

### 8. Azure OpenAI + Dynamics 365

```python
import openai

def classify_case_with_ai(client: Dynamics365Client, case_id: str,
                          aoai_client: openai.AzureOpenAI) -> str:
    """Classify a D365 case using Azure OpenAI and update the record."""
    case = client.get_by_id("incidents", case_id, select="title,description")

    classification = aoai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Classify the support case into: Billing, Technical, Account, General. Return only the category."},
            {"role": "user", "content": f"Title: {case['title']}\nDescription: {case.get('description', 'N/A')}"},
        ],
    )
    category = classification.choices[0].message.content.strip()
    client.update("incidents", case_id, {"new_aicategory": category})
    return category

def analyze_deal_risk(client: Dynamics365Client, opp_id: str,
                      aoai_client: openai.AzureOpenAI) -> str:
    """Analyze deal risk using Azure OpenAI."""
    opp = client.get_by_id("opportunities", opp_id,
                           select="name,estimatedvalue,closeprobability,estimatedclosedate,modifiedon")
    response = aoai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a sales operations analyst. Assess deal risk."},
            {"role": "user", "content": f"Opportunity: {opp['name']}, Value: ${opp.get('estimatedvalue', 0):,.2f}, "
                                        f"Probability: {opp.get('closeprobability', 'N/A')}%, "
                                        f"Close date: {opp.get('estimatedclosedate', 'N/A')}, "
                                        f"Last modified: {opp.get('modifiedon', 'N/A')}"},
        ],
    )
    return response.choices[0].message.content
```

### 9. Agent Tool Functions

```python
def search_accounts(account_name: str) -> str:
    """Search Dynamics 365 accounts by name.

    Args:
        account_name: Full or partial account name
    """
    client = Dynamics365Client(TENANT_ID, CLIENT_ID, CLIENT_SECRET, CRM_URL)
    accounts = client.get("accounts", params={
        "$select": "name,revenue,telephone1",
        "$filter": f"contains(name,'{account_name}')",
        "$top": 10,
    }).get("value", [])
    if not accounts:
        return f"No accounts found matching '{account_name}'."
    lines = [f"- {a['name']} (Revenue: ${a.get('revenue', 0):,.0f})" for a in accounts]
    return f"Found {len(accounts)} account(s):\n" + "\n".join(lines)

def get_pipeline_summary() -> str:
    """Get a summary of the open sales pipeline."""
    client = Dynamics365Client(TENANT_ID, CLIENT_ID, CLIENT_SECRET, CRM_URL)
    opps = client.get("opportunities", params={
        "$select": "name,estimatedvalue,closeprobability",
        "$filter": "statecode eq 0",
        "$orderby": "estimatedvalue desc",
    }).get("value", [])
    total = sum(o.get("estimatedvalue", 0) for o in opps)
    weighted = sum(o.get("estimatedvalue", 0) * o.get("closeprobability", 0) / 100 for o in opps)
    return (f"Pipeline: {len(opps)} open opportunities\n"
            f"Total value: ${total:,.0f}\n"
            f"Weighted value: ${weighted:,.0f}")
```

## Dynamics 365 Copilot Features

| Module | Copilot Capabilities |
|---|---|
| **Sales** | Record summarization, email drafting, meeting prep, lead research, pipeline insights |
| **Service** | Case summarization, KB article generation, similar case suggestions, AI routing |
| **Finance** | Account reconciliation agent, collections drafting, cash flow forecasting |
| **Supply Chain** | Demand forecasting, intelligent order management, inventory anomaly detection |

## Anti-Patterns

- Not using `$select` -- always specify fields to reduce payload and improve performance
- Ignoring `@odata.nextLink` -- results are capped at 5000 per page by default
- Using OData `$filter` for complex joins -- use FetchXML instead
- Hardcoding GUIDs -- use alternate keys where available (e.g., `accounts(accountnumber='A001')`)
- Skipping Application User setup -- client credentials flow requires a D365 Application User with security roles
- Sending raw D365 data to LLMs without field selection -- filter sensitive fields before AI processing
- Not handling F&O `dataAreaId` -- Finance & Operations requires legal entity on most write operations

## References

- [Dataverse Web API Reference](https://learn.microsoft.com/en-us/power-apps/developer/data-platform/webapi/reference/about)
- [Dynamics 365 F&O Data Entities](https://learn.microsoft.com/en-us/dynamics365/fin-ops-core/dev-itpro/data-entities/data-entities)
- [MSAL Python Documentation](https://learn.microsoft.com/en-us/entra/msal/python/)
- [FetchXML Reference](https://learn.microsoft.com/en-us/power-apps/developer/data-platform/use-fetchxml-construct-query)
- [Dynamics 365 Copilot](https://learn.microsoft.com/en-us/dynamics365/copilot/ai-get-started)

<!-- Source: .faos/custom/skills/integrations/dynamics-365/SKILL.md -->