API wrapper for Dynamics365CRM written in Python
npx @tessl/cli install tessl/pypi-dynamics365crm-python@1.0.0A Python API wrapper for Microsoft Dynamics 365 CRM API v9.0 that provides OAuth2 authentication and CRUD operations for CRM entities including contacts, accounts, opportunities, leads, and campaigns.
pip install dynamics365crm-pythonfrom dynamics365crm.client import Clientfrom dynamics365crm.client import Client
# Initialize with access token
client = Client(
domain="https://tenant_name.crmX.dynamics.com",
access_token="your_access_token"
)
# Or initialize for OAuth2 flow
client = Client(
domain="https://tenant_name.crmX.dynamics.com",
client_id="your_client_id",
client_secret="your_client_secret"
)
# Create a contact
contact_id = client.create_contact(
firstname="John",
lastname="Doe",
emailaddress1="john.doe@example.com"
)
# Get all contacts
contacts = client.get_contacts()
# Update a contact
client.update_contact(
id=contact_id,
middlename="Michael"
)
# Delete a contact
client.delete_contact(contact_id)The library is built around a single Client class that provides:
Initialize the Dynamics 365 CRM client with domain and authentication credentials.
class Client:
api_path = "api/data/v9.0"
def __init__(self, domain: str, client_id: str = None, client_secret: str = None, access_token: str = None):
"""
Initialize the Dynamics 365 CRM client.
Class Attributes:
- api_path: API path for Dynamics 365 Web API v9.0
Parameters:
- domain: The Dynamics 365 tenant domain URL
- client_id: Azure AD application client ID (for OAuth2)
- client_secret: Azure AD application client secret (for OAuth2)
- access_token: Direct access token for API requests
"""Manage OAuth2 authentication flow and access tokens for API requests.
def set_access_token(self, token: str):
"""
Sets the access token for API requests.
Parameters:
- token: Access token string
Raises:
- AssertionError: If token is None
"""
def build_authorization_url(self, tenant_id: str, redirect_uri: str, state: str) -> str:
"""
Generate OAuth2 authorization URL for user consent.
Parameters:
- tenant_id: Azure AD tenant ID or "common"
- redirect_uri: Callback URL for authorization response
- state: Unique state identifier for security
Returns:
- str: Authorization URL for user redirection
"""
def exchange_code(self, tenant_id: str, redirect_uri: str, code: str) -> dict:
"""
Exchange authorization code for access token.
Parameters:
- tenant_id: Azure AD tenant ID or "common"
- redirect_uri: Must match the redirect_uri used in authorization
- code: Authorization code from callback
Returns:
- dict: Token response containing access_token, refresh_token, etc.
"""
def refresh_access_token(self, tenant_id: str, refresh_token: str) -> dict:
"""
Refresh access token using refresh token.
Parameters:
- tenant_id: Azure AD tenant ID or "common"
- refresh_token: Refresh token from previous authentication
Returns:
- dict: New token response or error dict with "error" key
"""
def build_msal_client(self, tenant_id: str):
"""
Create MSAL ConfidentialClientApplication instance for OAuth2 operations.
Parameters:
- tenant_id: Azure AD tenant ID
Returns:
- msal.ConfidentialClientApplication: MSAL client instance configured for tenant
"""Low-level HTTP request methods for direct API interaction.
def make_request(self, method: str, endpoint: str, expand: str = None, filter: str = None, orderby: str = None, select: str = None, skip: str = None, top: str = None, data = None, json = None, **kwargs):
"""
Core method for making API requests with OData query parameters.
Parameters:
- method: HTTP method ("get", "post", "patch", "delete")
- endpoint: API endpoint path
- expand: OData expand parameter for related entities
- filter: OData filter parameter for querying
- orderby: OData orderby parameter for sorting
- select: OData select parameter for field selection
- skip: OData skip parameter for pagination
- top: OData top parameter for limiting results
- data: Raw request data
- json: JSON request data
- **kwargs: Additional request parameters
Returns:
- dict: Parsed API response
Raises:
- AssertionError: If domain or access_token is None
- Exception: For various HTTP error responses
"""
def parse_response(self, response):
"""
Parse HTTP response and handle error conditions.
Parameters:
- response: requests.Response object
Returns:
- dict: JSON response data
- str: Entity GUID for successful create operations
- bool: True for successful operations without response data
Raises:
- Exception: For HTTP error status codes (400, 401, 403, 404, 412, 413, 500, 501, 503)
"""Generic CRUD operations for any Dynamics 365 entity type using OData query parameters.
def get_data(self, type: str, expand: str = None, filter: str = None, orderby: str = None, select: str = None, skip: str = None, top: str = None, **kwargs) -> dict:
"""
Generic method to retrieve data for any entity type.
Parameters:
- type: Entity type name (e.g., "contacts", "accounts")
- expand: OData expand parameter for related entities
- filter: OData filter parameter for querying
- orderby: OData orderby parameter for sorting
- select: OData select parameter for field selection
- skip: OData skip parameter for pagination
- top: OData top parameter for limiting results
Returns:
- dict: API response with entity data
Raises:
- Exception: If type is None
"""
def create_data(self, type: str, **kwargs) -> str:
"""
Generic method to create data for any entity type.
Parameters:
- type: Entity type name
- **kwargs: Entity field values
Returns:
- str: Created entity GUID or True if successful
Raises:
- Exception: If type is None
"""
def update_data(self, type: str, id: str, **kwargs) -> bool:
"""
Generic method to update data for any entity type.
Parameters:
- type: Entity type name
- id: Entity GUID
- **kwargs: Updated field values
Returns:
- bool: True if successful
Raises:
- Exception: If type or id is None
"""
def delete_data(self, type: str, id: str) -> bool:
"""
Generic method to delete data for any entity type.
Parameters:
- type: Entity type name
- id: Entity GUID
Returns:
- bool: True if successful
Raises:
- Exception: If type or id is None
"""CRUD operations for contact entities in Dynamics 365.
def get_contacts(self, expand: str = None, filter: str = None, orderby: str = None, select: str = None, skip: str = None, top: str = None, **kwargs) -> dict:
"""
Retrieve contacts with optional OData query parameters.
Parameters:
- expand: Related entities to include
- filter: Filter criteria
- orderby: Sort order
- select: Fields to return
- skip: Number of records to skip
- top: Maximum number of records
Returns:
- dict: API response with contact data
"""
def create_contact(self, **kwargs) -> str:
"""
Create a new contact.
Parameters:
- **kwargs: Contact field values (firstname, lastname, emailaddress1, etc.)
Returns:
- str: Created contact GUID or True if successful
"""
def update_contact(self, id: str, **kwargs) -> bool:
"""
Update an existing contact.
Parameters:
- id: Contact GUID
- **kwargs: Updated field values
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""
def delete_contact(self, id: str) -> bool:
"""
Delete a contact by ID.
Parameters:
- id: Contact GUID
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""CRUD operations for account entities in Dynamics 365.
def get_accounts(self, expand: str = None, filter: str = None, orderby: str = None, select: str = None, skip: str = None, top: str = None, **kwargs) -> dict:
"""
Retrieve accounts with optional OData query parameters.
Parameters:
- expand: Related entities to include
- filter: Filter criteria
- orderby: Sort order
- select: Fields to return
- skip: Number of records to skip
- top: Maximum number of records
Returns:
- dict: API response with account data
"""
def create_account(self, **kwargs) -> str:
"""
Create a new account.
Parameters:
- **kwargs: Account field values (name, websiteurl, etc.)
Returns:
- str: Created account GUID or True if successful
"""
def update_account(self, id: str, **kwargs) -> bool:
"""
Update an existing account.
Parameters:
- id: Account GUID
- **kwargs: Updated field values
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""
def delete_account(self, id: str) -> bool:
"""
Delete an account by ID.
Parameters:
- id: Account GUID
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""CRUD operations for opportunity entities in Dynamics 365.
def get_opportunities(self, expand: str = None, filter: str = None, orderby: str = None, select: str = None, skip: str = None, top: str = None, **kwargs) -> dict:
"""
Retrieve opportunities with optional OData query parameters.
Parameters:
- expand: Related entities to include
- filter: Filter criteria
- orderby: Sort order
- select: Fields to return
- skip: Number of records to skip
- top: Maximum number of records
Returns:
- dict: API response with opportunity data
"""
def create_opportunity(self, **kwargs) -> str:
"""
Create a new opportunity.
Parameters:
- **kwargs: Opportunity field values (name, etc.)
Returns:
- str: Created opportunity GUID or True if successful
"""
def update_opportunity(self, id: str, **kwargs) -> bool:
"""
Update an existing opportunity.
Parameters:
- id: Opportunity GUID
- **kwargs: Updated field values
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""
def delete_opportunity(self, id: str) -> bool:
"""
Delete an opportunity by ID.
Parameters:
- id: Opportunity GUID
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""CRUD operations for lead entities in Dynamics 365.
def get_leads(self, expand: str = None, filter: str = None, orderby: str = None, select: str = None, skip: str = None, top: str = None, **kwargs) -> dict:
"""
Retrieve leads with optional OData query parameters.
Parameters:
- expand: Related entities to include
- filter: Filter criteria
- orderby: Sort order
- select: Fields to return
- skip: Number of records to skip
- top: Maximum number of records
Returns:
- dict: API response with lead data
"""
def create_lead(self, **kwargs) -> str:
"""
Create a new lead.
Parameters:
- **kwargs: Lead field values (fullname, subject, mobilephone, websiteurl, etc.)
Returns:
- str: Created lead GUID or True if successful
"""
def update_lead(self, id: str, **kwargs) -> bool:
"""
Update an existing lead.
Parameters:
- id: Lead GUID
- **kwargs: Updated field values
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""
def delete_lead(self, id: str) -> bool:
"""
Delete a lead by ID.
Parameters:
- id: Lead GUID
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""CRUD operations for campaign entities in Dynamics 365.
def get_campaigns(self, expand: str = None, filter: str = None, orderby: str = None, select: str = None, skip: str = None, top: str = None, **kwargs) -> dict:
"""
Retrieve campaigns with optional OData query parameters.
Parameters:
- expand: Related entities to include
- filter: Filter criteria
- orderby: Sort order
- select: Fields to return
- skip: Number of records to skip
- top: Maximum number of records
Returns:
- dict: API response with campaign data
"""
def create_campaign(self, **kwargs) -> str:
"""
Create a new campaign.
Parameters:
- **kwargs: Campaign field values (name, description, etc.)
Returns:
- str: Created campaign GUID or True if successful
"""
def update_campaign(self, id: str, **kwargs) -> bool:
"""
Update an existing campaign.
Parameters:
- id: Campaign GUID
- **kwargs: Updated field values
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""
def delete_campaign(self, id: str) -> bool:
"""
Delete a campaign by ID.
Parameters:
- id: Campaign GUID
Returns:
- bool: True if successful
Raises:
- Exception: If id is empty
"""The client automatically handles various HTTP error responses:
All error conditions raise Exception with descriptive messages including the URL, status code, and raw error response.
from dynamics365crm.client import Client
# Initialize client for OAuth2
client = Client(
"https://tenant_name.crmX.dynamics.com",
client_id="your_client_id",
client_secret="your_client_secret"
)
# Step 1: Get authorization URL
auth_url = client.build_authorization_url(
tenant_id="your_tenant_id",
redirect_uri="https://yourapp.com/callback",
state="unique_state_string"
)
print(f"Visit: {auth_url}")
# Step 2: Exchange code for token (after user consent)
token_response = client.exchange_code(
tenant_id="your_tenant_id",
redirect_uri="https://yourapp.com/callback",
code="authorization_code_from_callback"
)
# Step 3: Set access token
client.set_access_token(token_response['access_token'])
# Step 4: Use the client for API calls
contacts = client.get_contacts()# Create a contact with multiple fields
contact_id = client.create_contact(
firstname="Jane",
lastname="Smith",
middlename="Elizabeth",
emailaddress1="jane.smith@company.com",
mobilephone="555-0123",
jobtitle="Sales Manager"
)
# Get contacts with filtering and selection
filtered_contacts = client.get_contacts(
filter="contains(lastname,'Smith')",
select="fullname,emailaddress1,jobtitle",
orderby="lastname",
top="10"
)
# Update contact information
client.update_contact(
id=contact_id,
jobtitle="Senior Sales Manager",
emailaddress1="jane.smith@newcompany.com"
)# Get accounts with expanded contact information
accounts_with_contacts = client.get_accounts(
expand="primarycontactid($select=fullname,emailaddress1)",
select="name,websiteurl,telephone1"
)
# Create related entities
account_id = client.create_account(
name="Tech Solutions Inc",
websiteurl="https://techsolutions.com",
telephone1="555-0100"
)
opportunity_id = client.create_opportunity(
name="Software Implementation",
description="CRM software implementation project",
estimatedvalue=50000
)