frank-luongt/skills/codex/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.

Platform Note: This skill was designed for multi-agent execution. In Codex, treat sub-agent instructions as sequential steps to complete thoroughly within a single agent context.

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)

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

# 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

# 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,
        "[email protected]": 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

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)

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)

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

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

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

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
• Dynamics 365 F&O Data Entities
• MSAL Python Documentation
• FetchXML Reference
• Dynamics 365 Copilot

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