Automate Microsoft 365 User Management

Create, update, and manage Microsoft 365 users with Microsoft Graph API

Automate common Microsoft 365 user management tasks using Microsoft Graph API. This guide shows practical patterns for MSP workflows.

Prerequisites

OAuth connection configured for Microsoft Graph
Required scopes: User.ReadWrite.All, Group.ReadWrite.All

Basic Pattern

All Microsoft Graph workflows follow the same structure:

from bifrost import workflow, oauth, ExecutionContext
import aiohttp

@workflow(name="graph_workflow")
async def graph_workflow(context: ExecutionContext):
    # Get OAuth token
    token_data = await oauth.get_token("microsoft")
    if not token_data:
        raise ValueError("Microsoft Graph OAuth not configured")

    # Make API call
    url = "https://graph.microsoft.com/v1.0/users"
    headers = {"Authorization": f"Bearer {token_data['access_token']}"}

    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as resp:
            resp.raise_for_status()
            return await resp.json()

List Users

Get all users in the tenant:

from bifrost import workflow, oauth, ExecutionContext
import aiohttp

@workflow(name="list_users", description="Get all users")
async def list_users(context: ExecutionContext):
    token_data = await oauth.get_token("microsoft")

    url = "https://graph.microsoft.com/v1.0/users"
    headers = {"Authorization": f"Bearer {token_data['access_token']}"}

    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as resp:
            resp.raise_for_status()
            data = await resp.json()
            return {"users": data.get("value", [])}

Create New User

Provision a new Microsoft 365 user:

from bifrost import workflow, param, oauth, ExecutionContext
import aiohttp

@workflow(name="create_user", description="Create new M365 user")
@param("email", type="email", required=True)
@param("display_name", type="string", required=True)
@param("password", type="string", required=True)
async def create_user(
    context: ExecutionContext,
    email: str,
    display_name: str,
    password: str
):
    token_data = await oauth.get_token("microsoft")

    url = "https://graph.microsoft.com/v1.0/users"
    headers = {
        "Authorization": f"Bearer {token_data['access_token']}",
        "Content-Type": "application/json"
    }

    payload = {
        "accountEnabled": True,
        "displayName": display_name,
        "mailNickname": email.split("@")[0],
        "userPrincipalName": email,
        "passwordProfile": {
            "forceChangePasswordNextSignIn": True,
            "password": password
        }
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, json=payload) as resp:
            if resp.status == 201:
                user = await resp.json()
                return {"success": True, "user_id": user["id"]}
            else:
                error = await resp.text()
                return {"success": False, "error": error}

Update User Properties

Modify existing user attributes:

@workflow(name="update_user", description="Update user properties")
@param("user_id", type="string", required=True)
@param("job_title", type="string", required=False)
@param("department", type="string", required=False)
async def update_user(
    context: ExecutionContext,
    user_id: str,
    job_title: str = None,
    department: str = None
):
    token_data = await oauth.get_token("microsoft")

    url = f"https://graph.microsoft.com/v1.0/users/{user_id}"
    headers = {
        "Authorization": f"Bearer {token_data['access_token']}",
        "Content-Type": "application/json"
    }

    # Build payload with only provided values
    payload = {}
    if job_title:
        payload["jobTitle"] = job_title
    if department:
        payload["department"] = department

    async with aiohttp.ClientSession() as session:
        async with session.patch(url, headers=headers, json=payload) as resp:
            if resp.status == 204:
                return {"success": True}
            else:
                return {"success": False, "error": await resp.text()}

Add User to Group

Add users to security or Microsoft 365 groups:

@workflow(name="add_to_group", description="Add user to group")
@param("user_id", type="string", required=True)
@param("group_id", type="string", required=True)
async def add_to_group(context: ExecutionContext, user_id: str, group_id: str):
    token_data = await oauth.get_token("microsoft")

    url = f"https://graph.microsoft.com/v1.0/groups/{group_id}/members/$ref"
    headers = {
        "Authorization": f"Bearer {token_data['access_token']}",
        "Content-Type": "application/json"
    }

    payload = {
        "@odata.id": f"https://graph.microsoft.com/v1.0/users/{user_id}"
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, json=payload) as resp:
            if resp.status == 204:
                return {"success": True}
            else:
                return {"success": False, "error": await resp.text()}

Assign License

Assign Microsoft 365 licenses to users:

@workflow(name="assign_license", description="Assign M365 license")
@param("user_id", type="string", required=True)
@param("sku_id", type="string", required=True)
async def assign_license(context: ExecutionContext, user_id: str, sku_id: str):
    token_data = await oauth.get_token("microsoft")

    url = f"https://graph.microsoft.com/v1.0/users/{user_id}/assignLicense"
    headers = {
        "Authorization": f"Bearer {token_data['access_token']}",
        "Content-Type": "application/json"
    }

    payload = {
        "addLicenses": [{"skuId": sku_id}],
        "removeLicenses": []
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, json=payload) as resp:
            if resp.status == 200:
                return {"success": True}
            else:
                return {"success": False, "error": await resp.text()}

Handle Pagination

Get all results from large datasets:

@workflow(name="get_all_users", description="Get all users with pagination")
async def get_all_users(context: ExecutionContext):
    token_data = await oauth.get_token("microsoft")
    headers = {"Authorization": f"Bearer {token_data['access_token']}"}

    all_users = []
    url = "https://graph.microsoft.com/v1.0/users?$top=100"

    async with aiohttp.ClientSession() as session:
        while url:
            async with session.get(url, headers=headers) as resp:
                resp.raise_for_status()
                data = await resp.json()
                all_users.extend(data.get("value", []))

                # Get next page URL
                url = data.get("@odata.nextLink")

    return {"users": all_users, "count": len(all_users)}

Filter and Select

Use OData queries to get specific data:

url = "https://graph.microsoft.com/v1.0/users?$filter=department eq 'Sales'"

url = "https://graph.microsoft.com/v1.0/users?$select=displayName,mail,jobTitle"

url = "https://graph.microsoft.com/v1.0/users?$filter=department eq 'Sales'&$select=displayName,mail&$top=50"

Error Handling

Handle common Graph API errors:

@workflow(name="safe_get_user")
@param("user_id", type="string", required=True)
async def safe_get_user(context: ExecutionContext, user_id: str):
    token_data = await oauth.get_token("microsoft")
    if not token_data:
        return {"success": False, "error": "OAuth not configured"}

    url = f"https://graph.microsoft.com/v1.0/users/{user_id}"
    headers = {"Authorization": f"Bearer {token_data['access_token']}"}

    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as resp:
            if resp.status == 200:
                user = await resp.json()
                return {"success": True, "user": user}
            elif resp.status == 404:
                return {"success": False, "error": "User not found"}
            elif resp.status == 403:
                return {"success": False, "error": "Insufficient permissions"}
            else:
                error = await resp.text()
                return {"success": False, "error": error}

Common Scopes

Operation	Required Scope
Read users	`User.Read.All`
Create/update users	`User.ReadWrite.All`
Read groups	`Group.Read.All`
Manage group membership	`GroupMember.ReadWrite.All`
Read licenses	`Organization.Read.All`
Assign licenses	`Organization.ReadWrite.All`

Best Practices

Always check token: Verify oauth.get_token() doesn’t return None
Use raise_for_status(): Let HTTP errors bubble as exceptions
Select only needed fields: Use $select to reduce payload size
Handle pagination: Large result sets require paging
Respect rate limits: Graph API has throttling - implement retries if needed

Next Steps

OAuth Integration - Set up Microsoft Graph OAuth
Store API Keys Securely - Manage credentials
Custom APIs - Integrate any REST API