Microsoft Licenses

Send Microsoft 365 plans, accounts, licenses, and app usage to Viio with the Python SDK, including service plans and subscriptions.

Last updated October 15, 2026

This example sends your Microsoft 365 licensing and usage to Viio, the same license and Microsoft 365 usage data that Viio's managed Microsoft integration sends:

Record	Microsoft Graph source	Sent as
Plan	Each subscribed SKU, with its service plans and the company subscriptions behind it	`MicrosoftPlanDetails`
Account	Each user	`GenericAccountDetails`
License	Each SKU assigned to a user, directly or through a group	`GenericLicenseDetails`
Usage	Each service a user was last active in, such as Exchange, OneDrive, SharePoint, or Teams, from the Microsoft 365 active users report	`GenericUsageDetails`
Usage	Each app a user used, such as Outlook, Word, or Excel, from the Microsoft 365 Apps usage report	`GenericUsageDetails`

Plans carry the SKU's seat counts and, for each subscription, its renewal date, trial flag, and total licenses. Usage covers the last 180 days.

For SDK installation and environment variables, see Shared Setup. MicrosoftPlanDetails requires SDK 0.5.0 or later.

The script must process every Microsoft Graph page before it completes the batch. Completing a partial snapshot marks valid plans, accounts, or licenses as removed.

Microsoft Graph Permissions

Grant Organization.Read.All, User.Read.All, and Reports.Read.All:

Endpoint	Data	Least privileged application permission
`GET /subscribedSkus`	SKUs, seat counts, and service plans	`LicenseAssignment.Read.All`
`GET /directory/subscriptions`	Company subscriptions	`Organization.Read.All`
`GET /users`	Users and their assigned licenses	`User.Read.All`
`GET /reports/getOffice365ActiveUserDetail`	Each user's last activity date per service	`Reports.Read.All`
`GET /reports/getM365AppUserDetail`	The Microsoft 365 apps each user used	`Reports.Read.All`

Organization.Read.All also covers /subscribedSkus. If the application has LicenseAssignment.Read.All instead, the subscriptions request returns 403 Forbidden, and the script sends the plans without subscriptions. Without Reports.Read.All, the report requests return 403 Forbidden, and the script sends no usage.

Microsoft 365 hides usernames in usage reports by default, and that includes the Microsoft Graph reports. The script matches report rows to users by user principal name, so it skips every row while names are hidden. To send usage, a Global Administrator must clear Conceal user, group, and site names in all reports in the Microsoft 365 admin center under Settings > Org settings > Services > Reports. The change takes a few minutes to apply.

Script

Create microsoft_licenses.py:

import argparse
import asyncio
import csv
import io
import json
import logging
import os
from collections import defaultdict
from collections.abc import Awaitable, Callable
from typing import Any
from urllib.error import HTTPError
from urllib.parse import urlencode, urlsplit
from urllib.request import Request, urlopen

from google.protobuf.json_format import MessageToJson
from google.protobuf.timestamp_pb2 import Timestamp
from viio_sync_api import SyncApiAuthentication, SyncApiClient, SyncApiClientOptions
from viio_sync_api.v1 import (
    Account,
    AccountRecords,
    GenericAccountDetails,
    GenericLicenseDetails,
    GenericUsageDetails,
    License,
    LicenseRecords,
    MicrosoftCompanySubscription,
    MicrosoftLicenseUnitsDetail,
    MicrosoftPlanDetails,
    MicrosoftServicePlanInfo,
    Plan,
    PlanRecords,
    SyncRecordsRequest,
    Usage,
    UsageRecords,
)

Write = Callable[[SyncRecordsRequest], Awaitable[None]]
ReportRow = dict[str, str]

GRAPH_SUBSCRIBED_SKUS_URL = "https://graph.microsoft.com/v1.0/subscribedSkus"
GRAPH_SUBSCRIPTIONS_URL = "https://graph.microsoft.com/v1.0/directory/subscriptions"
GRAPH_USERS_URL = "https://graph.microsoft.com/v1.0/users?" + urlencode(
    {
        "$select": "id,displayName,mail,userPrincipalName,accountEnabled,userType,assignedLicenses",
        "$top": "999",
    }
)
GRAPH_ACTIVE_USERS_REPORT_URL = (
    "https://graph.microsoft.com/v1.0/reports/getOffice365ActiveUserDetail(period='D180')"
)
GRAPH_APPS_REPORT_URL = "https://graph.microsoft.com/v1.0/reports/getM365AppUserDetail(period='D180')"
SKIPPED_CAPABILITY_STATUSES = {"Suspended", "Deleted", "LockedOut"}
LAST_ACTIVITY_COLUMN_SUFFIX = " Last Activity Date"
APPS_REPORT_NON_APP_COLUMNS = {
    "Report Refresh Date",
    "User Principal Name",
    "Last Activation Date",
    "Last Activity Date",
    "Report Period",
    "Windows",
    "Mac",
    "Mobile",
    "Web",
}
USAGE_RECORDS_PER_WRITE = 1000

logger = logging.getLogger("microsoft_licenses")


def request_json(request: Request) -> dict[str, Any]:
    with urlopen(request, timeout=30) as response:
        payload = json.load(response)
    if not isinstance(payload, dict):
        raise RuntimeError("Remote service returned an invalid JSON response")
    return payload


def request_graph_token() -> str:
    tenant_id = os.environ["MICROSOFT_TENANT_ID"]
    allowed_characters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-"
    if not tenant_id or any(character not in allowed_characters for character in tenant_id):
        raise ValueError("MICROSOFT_TENANT_ID must be a tenant ID or verified domain name")

    request = Request(
        f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
        data=urlencode(
            {
                "grant_type": "client_credentials",
                "client_id": os.environ["MICROSOFT_CLIENT_ID"],
                "client_secret": os.environ["MICROSOFT_CLIENT_SECRET"],
                "scope": "https://graph.microsoft.com/.default",
            }
        ).encode(),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    token = request_json(request).get("access_token")
    if not isinstance(token, str) or not token:
        raise RuntimeError("Microsoft identity platform returned no access token")
    return token


def request_graph_page(url: str, access_token: str) -> tuple[list[dict[str, Any]], str | None]:
    parsed_url = urlsplit(url)
    if parsed_url.scheme != "https" or parsed_url.hostname != "graph.microsoft.com":
        raise RuntimeError("Microsoft Graph returned an invalid pagination URL")

    payload = request_json(
        Request(
            url,
            headers={
                "Accept": "application/json",
                "Authorization": f"Bearer {access_token}",
            },
        )
    )
    items = payload.get("value")
    if not isinstance(items, list) or not all(isinstance(item, dict) for item in items):
        raise RuntimeError("Microsoft Graph returned an invalid collection response")

    next_url = payload.get("@odata.nextLink")
    if next_url is not None and not isinstance(next_url, str):
        raise RuntimeError("Microsoft Graph returned an invalid pagination URL")
    return items, next_url


def request_graph_collection(url: str, access_token: str) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    next_url: str | None = url
    while next_url is not None:
        page, next_url = request_graph_page(next_url, access_token)
        items.extend(page)
    return items


def request_graph_report(url: str, access_token: str) -> list[ReportRow]:
    request = Request(url)
    request.add_unredirected_header("Authorization", f"Bearer {access_token}")
    with urlopen(request, timeout=300) as response:
        content = response.read().decode("utf-8-sig")
    return list(csv.DictReader(io.StringIO(content)))


def required_string(source: dict[str, Any], field: str) -> str:
    value = source.get(field)
    if not isinstance(value, str) or not value:
        raise RuntimeError(f"Microsoft Graph returned an object without {field}")
    return value


def optional_string(source: dict[str, Any], field: str) -> str | None:
    value = source.get(field)
    return value if isinstance(value, str) and value else None


def optional_int(source: dict[str, Any], field: str) -> int | None:
    value = source.get(field)
    return value if isinstance(value, int) and not isinstance(value, bool) else None


def optional_bool(source: dict[str, Any], field: str) -> bool | None:
    value = source.get(field)
    return value if isinstance(value, bool) else None


def optional_timestamp(source: dict[str, Any], field: str) -> Timestamp | None:
    value = optional_string(source, field)
    if value is None:
        return None
    timestamp = Timestamp()
    timestamp.FromJsonString(value)
    return timestamp


def strings(source: dict[str, Any], field: str) -> list[str]:
    value = source.get(field)
    return [item for item in value if isinstance(item, str)] if isinstance(value, list) else []


def objects(source: dict[str, Any], field: str) -> list[dict[str, Any]]:
    value = source.get(field)
    return [item for item in value if isinstance(item, dict)] if isinstance(value, list) else []


def service_plan_info(service_plan: dict[str, Any]) -> MicrosoftServicePlanInfo:
    return MicrosoftServicePlanInfo(
        service_plan_id=optional_string(service_plan, "servicePlanId"),
        service_plan_name=optional_string(service_plan, "servicePlanName"),
        provisioning_status=optional_string(service_plan, "provisioningStatus"),
        applies_to=optional_string(service_plan, "appliesTo"),
    )


def company_subscription(subscription: dict[str, Any]) -> MicrosoftCompanySubscription:
    return MicrosoftCompanySubscription(
        id=required_string(subscription, "id"),
        commerce_subscription_id=optional_string(subscription, "commerceSubscriptionId"),
        created_date_time=optional_timestamp(subscription, "createdDateTime"),
        is_trial=optional_bool(subscription, "isTrial"),
        next_lifecycle_date_time=optional_timestamp(subscription, "nextLifecycleDateTime"),
        owner_id=optional_string(subscription, "ownerId"),
        owner_tenant_id=optional_string(subscription, "ownerTenantId"),
        owner_type=optional_string(subscription, "ownerType"),
        service_status=[service_plan_info(item) for item in objects(subscription, "serviceStatus")],
        sku_id=optional_string(subscription, "skuId"),
        sku_part_number=optional_string(subscription, "skuPartNumber"),
        status=optional_string(subscription, "status"),
        total_licenses=optional_int(subscription, "totalLicenses"),
    )


def prepaid_units(sku: dict[str, Any]) -> MicrosoftLicenseUnitsDetail | None:
    units = sku.get("prepaidUnits")
    if not isinstance(units, dict):
        return None
    return MicrosoftLicenseUnitsDetail(
        enabled=optional_int(units, "enabled"),
        locked_out=optional_int(units, "lockedOut"),
        suspended=optional_int(units, "suspended"),
        warning=optional_int(units, "warning"),
    )


def plan_from_subscribed_sku(
    sku: dict[str, Any],
    subscriptions_by_sku_id: dict[str, list[MicrosoftCompanySubscription]],
) -> Plan:
    sku_id = required_string(sku, "skuId")
    return Plan(
        microsoft=MicrosoftPlanDetails(
            sku_id=sku_id,
            sku_part_number=required_string(sku, "skuPartNumber"),
            account_id=optional_string(sku, "accountId"),
            account_name=optional_string(sku, "accountName"),
            applies_to=optional_string(sku, "appliesTo"),
            capability_status=optional_string(sku, "capabilityStatus"),
            consumed_units=optional_int(sku, "consumedUnits"),
            prepaid_units=prepaid_units(sku),
            service_plans=[service_plan_info(item) for item in objects(sku, "servicePlans")],
            subscription_ids=strings(sku, "subscriptionIds"),
            subscriptions=subscriptions_by_sku_id.get(sku_id.lower(), []),
        )
    )


def fetch_company_subscriptions(access_token: str) -> dict[str, list[MicrosoftCompanySubscription]]:
    try:
        subscriptions = request_graph_collection(GRAPH_SUBSCRIPTIONS_URL, access_token)
    except HTTPError as error:
        if error.code != 403:
            raise
        logger.warning("Organization.Read.All is not granted, so plans are sent without subscriptions")
        return {}

    subscriptions_by_sku_id: dict[str, list[MicrosoftCompanySubscription]] = defaultdict(list)
    for subscription in subscriptions:
        sku_id = optional_string(subscription, "skuId")
        if sku_id is not None:
            subscriptions_by_sku_id[sku_id.lower()].append(company_subscription(subscription))
    return subscriptions_by_sku_id


def fetch_plans(access_token: str) -> list[Plan]:
    subscriptions_by_sku_id = fetch_company_subscriptions(access_token)
    skus = request_graph_collection(GRAPH_SUBSCRIBED_SKUS_URL, access_token)
    return [
        plan_from_subscribed_sku(sku, subscriptions_by_sku_id)
        for sku in skus
        if sku.get("capabilityStatus") not in SKIPPED_CAPABILITY_STATUSES
    ]


def account_from_graph_user(user: dict[str, Any]) -> Account:
    account_enabled = optional_bool(user, "accountEnabled")
    return Account(
        generic=GenericAccountDetails(
            id=required_string(user, "id"),
            name=optional_string(user, "displayName"),
            email=optional_string(user, "mail") or optional_string(user, "userPrincipalName"),
            active=True if account_enabled is None else account_enabled,
            type=optional_string(user, "userType"),
        )
    )


def licenses_from_graph_user(user: dict[str, Any], plan_ids: dict[str, str]) -> list[License]:
    user_id = required_string(user, "id")
    licenses: list[License] = []
    for assigned_license in objects(user, "assignedLicenses"):
        sku_id = optional_string(assigned_license, "skuId")
        plan_id = plan_ids.get(sku_id.lower()) if sku_id is not None else None
        if plan_id is not None:
            licenses.append(
                License(
                    generic=GenericLicenseDetails(
                        source_id=f"{user_id}_{plan_id}",
                        source_account_id=user_id,
                        source_plan_id=plan_id,
                        billable=True,
                    )
                )
            )
    return licenses


def usage_record(user_id: str, product_name: str, last_activity_date: str) -> Usage:
    last_activity = Timestamp()
    last_activity.FromJsonString(f"{last_activity_date}T00:00:00Z")
    return Usage(
        generic=GenericUsageDetails(
            id=f"{user_id}-{product_name}-{last_activity_date}",
            source_actor_id=user_id,
            external_product_name=product_name,
            last_activity=last_activity,
        )
    )


def usage_from_active_users_row(row: ReportRow, user_id: str) -> list[Usage]:
    return [
        usage_record(user_id, column.removesuffix(LAST_ACTIVITY_COLUMN_SUFFIX), last_activity_date)
        for column, last_activity_date in row.items()
        if column.endswith(LAST_ACTIVITY_COLUMN_SUFFIX) and last_activity_date
    ]


def usage_from_apps_row(row: ReportRow, user_id: str) -> list[Usage]:
    last_activity_date = row.get("Last Activity Date")
    if not last_activity_date:
        return []
    return [
        usage_record(user_id, column, last_activity_date)
        for column, used in row.items()
        if column not in APPS_REPORT_NON_APP_COLUMNS and used == "Yes"
    ]


def usage_from_report(
    rows: list[ReportRow],
    user_ids: dict[str, str],
    usage_from_row: Callable[[ReportRow, str], list[Usage]],
) -> list[Usage]:
    usage: list[Usage] = []
    skipped_rows = 0
    for row in rows:
        user_id = user_ids.get((row.get("User Principal Name") or "").lower())
        if user_id is None:
            skipped_rows += 1
        else:
            usage.extend(usage_from_row(row, user_id))
    if skipped_rows:
        logger.warning("Skipped %d report rows that match no user", skipped_rows)
    return usage


def fetch_usage(access_token: str, user_ids: dict[str, str]) -> list[Usage]:
    try:
        active_users_rows = request_graph_report(GRAPH_ACTIVE_USERS_REPORT_URL, access_token)
        apps_rows = request_graph_report(GRAPH_APPS_REPORT_URL, access_token)
    except HTTPError as error:
        if error.code != 403:
            raise
        logger.warning("Reports.Read.All is not granted, so no usage is sent")
        return []
    return [
        *usage_from_report(active_users_rows, user_ids, usage_from_active_users_row),
        *usage_from_report(apps_rows, user_ids, usage_from_apps_row),
    ]


async def sync_microsoft_licenses(write: Write) -> dict[str, int]:
    access_token = await asyncio.to_thread(request_graph_token)
    plans = await asyncio.to_thread(fetch_plans, access_token)
    await write(SyncRecordsRequest(plans=PlanRecords(records=plans)))
    plan_ids = {plan.microsoft.sku_id.lower(): plan.microsoft.sku_id for plan in plans}

    counts = {"plans": len(plans), "accounts": 0, "licenses": 0, "usage": 0}
    user_ids: dict[str, str] = {}
    next_url: str | None = GRAPH_USERS_URL
    while next_url is not None:
        users, next_url = await asyncio.to_thread(request_graph_page, next_url, access_token)
        for user in users:
            principal_name = optional_string(user, "userPrincipalName")
            if principal_name is not None:
                user_ids[principal_name.lower()] = required_string(user, "id")

        accounts = [account_from_graph_user(user) for user in users]
        if accounts:
            await write(SyncRecordsRequest(accounts=AccountRecords(records=accounts)))
            counts["accounts"] += len(accounts)

        licenses = [record for user in users for record in licenses_from_graph_user(user, plan_ids)]
        if licenses:
            await write(SyncRecordsRequest(licenses=LicenseRecords(records=licenses)))
            counts["licenses"] += len(licenses)

    if counts["accounts"] == 0:
        await write(SyncRecordsRequest(accounts=AccountRecords()))
    if counts["licenses"] == 0:
        await write(SyncRecordsRequest(licenses=LicenseRecords()))

    usage = await asyncio.to_thread(fetch_usage, access_token, user_ids)
    for start in range(0, len(usage), USAGE_RECORDS_PER_WRITE):
        await write(SyncRecordsRequest(usage=UsageRecords(records=usage[start : start + USAGE_RECORDS_PER_WRITE])))
    counts["usage"] = len(usage)
    return counts


async def run_batch(sync: Callable[[Write], Awaitable[dict[str, int]]]) -> dict[str, int]:
    options = SyncApiClientOptions(
        endpoint=os.environ["VIIO_SYNC_API_ENDPOINT"],
        authentication=SyncApiAuthentication.api_key(os.environ["VIIO_SYNC_API_KEY"]),
    )

    async with SyncApiClient(options) as client:
        batch = await client.start_batch(os.environ["VIIO_DIRECT_INTEGRATION_INSTALLATION_ID"])
        try:
            counts = await sync(batch.write)
            await batch.complete()
        except BaseException:
            await batch.abort()
            raise
    return counts


async def print_request(request: SyncRecordsRequest) -> None:
    print(MessageToJson(request))


async def main() -> None:
    parser = argparse.ArgumentParser(description="Sync Microsoft 365 plans, accounts, licenses, and usage to Viio")
    parser.add_argument("--dry-run", action="store_true", help="print the records instead of sending them")
    arguments = parser.parse_args()

    counts = await (sync_microsoft_licenses(print_request) if arguments.dry_run else run_batch(sync_microsoft_licenses))
    logger.info("Synced %s", counts)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

The script follows the same rules as Viio's managed Microsoft integration:

It skips SKUs whose capabilityStatus is Suspended, Deleted, or LockedOut, and doesn't send licenses for them. Viio removes those plans when the batch completes.
It matches subscriptions to SKUs by skuId, ignoring case.
It uses the SKU ID as the plan ID, and {userId}_{skuId} as the license ID, so later runs update the same records.
It sends the plan's Graph values unchanged. Viio derives the plan name from skuPartNumber and the seat counts from consumedUnits and prepaidUnits.
It uses the user's mail as the account email, or userPrincipalName when mail is empty.
It matches report rows to users by userPrincipalName, ignoring case, and skips rows without a match, such as rows for deleted users.
From the active users report, it sends one usage record for each service with a last activity date. The product name is the column name without Last Activity Date, such as Exchange or Teams.
From the Microsoft 365 Apps report, it sends one usage record for each app column marked Yes, including platform-specific columns such as Outlook (Windows). It skips the Windows, Mac, Mobile, and Web platform columns. This report has one last activity date per user, so every app gets that date.
It uses {userId}-{product}-{lastActivityDate} as the usage ID. Viio publishes only activity that is newer than what it already has for the account and product, so resending the 180-day reports on every run doesn't create duplicates.

The plans go in one write. Each Graph page of up to 999 users becomes one accounts write and one licenses write. When a snapshot has no records, the script sends an empty wrapper, so Viio records it as intentionally empty. Usage follows the snapshots in writes of up to 1,000 records. Usage is event-based, so the script sends no usage write when the reports have no activity.

Microsoft Graph answers each report request with a redirect to a preauthenticated download URL. The script sets the Authorization header with add_unredirected_header, so the access token isn't sent to the download host.

Run the Script

Preview the records without contacting Viio:

python microsoft_licenses.py --dry-run

The dry run prints each request as JSON. For one SKU and one user, the output looks like this, with the usage write shortened to two records:

{
  "plans": {
    "records": [
      {
        "microsoft": {
          "skuId": "c7df2760-2c81-4ef7-b578-5b5392b571df",
          "skuPartNumber": "ENTERPRISEPREMIUM",
          "accountName": "Contoso",
          "appliesTo": "User",
          "capabilityStatus": "Enabled",
          "consumedUnits": 14,
          "prepaidUnits": {
            "enabled": 25,
            "lockedOut": 0,
            "suspended": 0,
            "warning": 0
          },
          "servicePlans": [
            {
              "servicePlanId": "8c098270-9dd4-4350-9b30-ba4703f3b36b",
              "servicePlanName": "ADALLOM_S_O365",
              "provisioningStatus": "Success",
              "appliesTo": "Company"
            }
          ],
          "subscriptionIds": [
            "43d26afe-cb98-48b9-acc4-ae3ef2ac6c51"
          ],
          "subscriptions": [
            {
              "id": "43d26afe-cb98-48b9-acc4-ae3ef2ac6c51",
              "commerceSubscriptionId": "f9c1ea2d-2c6e-4717-8c3b-7130812d70ba",
              "createdDateTime": "2025-01-01T00:00:00Z",
              "isTrial": false,
              "nextLifecycleDateTime": "2027-01-01T00:00:00Z",
              "skuId": "c7df2760-2c81-4ef7-b578-5b5392b571df",
              "skuPartNumber": "ENTERPRISEPREMIUM",
              "status": "Enabled",
              "totalLicenses": 25
            }
          ]
        }
      }
    ]
  }
}
{
  "accounts": {
    "records": [
      {
        "generic": {
          "id": "87d349ed-44d7-43e1-9a83-5f2406dee5bd",
          "name": "Adele Vance",
          "email": "adelev@contoso.com",
          "active": true,
          "type": "Member"
        }
      }
    ]
  }
}
{
  "licenses": {
    "records": [
      {
        "generic": {
          "sourceAccountId": "87d349ed-44d7-43e1-9a83-5f2406dee5bd",
          "sourcePlanId": "c7df2760-2c81-4ef7-b578-5b5392b571df",
          "billable": true,
          "sourceId": "87d349ed-44d7-43e1-9a83-5f2406dee5bd_c7df2760-2c81-4ef7-b578-5b5392b571df"
        }
      }
    ]
  }
}
{
  "usage": {
    "records": [
      {
        "generic": {
          "id": "87d349ed-44d7-43e1-9a83-5f2406dee5bd-Exchange-2026-10-12",
          "sourceActorId": "87d349ed-44d7-43e1-9a83-5f2406dee5bd",
          "externalProductName": "Exchange",
          "lastActivity": "2026-10-12T00:00:00Z"
        }
      },
      {
        "generic": {
          "id": "87d349ed-44d7-43e1-9a83-5f2406dee5bd-Outlook-2026-10-13",
          "sourceActorId": "87d349ed-44d7-43e1-9a83-5f2406dee5bd",
          "externalProductName": "Outlook",
          "lastActivity": "2026-10-13T00:00:00Z"
        }
      }
    ]
  }
}

Send the snapshot and usage to Viio:

python microsoft_licenses.py

If a Microsoft Graph or Sync API call fails, the script aborts the batch and raises the error. Aborting doesn't remove any records, because Viio reconciles snapshots only when a batch completes. See the quickstart for the retry sequence.

Next: run the syncs on AWS Lambda Browse plan and license models Browse usage models