Fix Graph 401 The server returned an unexpected status code

We are trying to use the Graph API to send emails automatically through a no-reply account. We have followed the Microsoft guide on https://learn.microsoft.com/en-us/graph/tutorials/python-email to the tee, and have tried multiple other python approaches. However, each time we get:

kiota_abstractions.api_error.APIError:

APIError

Code: 401

message: The server returned an unexpected status code and the error registered for this code failed to deserialize: <class 'NoneType'>

The same goes for trying to manage graph permissions through Connect-ExchangeOnline - each time, we are unable to get access, yielding

UnAuthorized

At C:\Program

Files\WindowsPowerShell\Modules\ExchangeOnlineManagement\3.9.0\netFramework\ExchangeOnlineManagement.psm1:770 char:21

+ throw $_.Exception;

+ ~~~~~~~~~~~~~~~~~~

+ CategoryInfo : OperationStopped: (:) [], UnauthorizedAccessException

+ FullyQualifiedErrorId : UnAuthorized

Is it possible that our Azure subscription does not have this enabled? Is there anything we can do from our end?

I have added the scripts we tried using. We are unable to use option 2 - List my inbox - clearly indicating that something is wrong:

main.py

import configparser
from msgraph.generated.models.o_data_errors.o_data_error import ODataError
from graph import Graph
import asyncio


async def main():
    print("Python Graph Tutorial\n")

    # Load settings
    config = configparser.ConfigParser()
    config.read(["config.cfg", "config.dev.cfg"])
    azure_settings = config["azure"]

    graph: Graph = Graph(azure_settings)

    await greet_user(graph)

    choice = -1

    while choice != 0:
        print("Please choose one of the following options:")
        print("0. Exit")
        print("1. Display access token")
        print("2. List my inbox")
        print("3. Send mail")
        print("4. Make a Graph call")

        try:
            choice = int(input())
        except ValueError:
            choice = -1

        try:
            if choice == 0:
                print("Goodbye...")
            elif choice == 1:
                await display_access_token(graph)
            elif choice == 2:
                await list_inbox(graph)
            elif choice == 3:
                await send_mail(graph)
            elif choice == 4:
                await make_graph_call(graph)
            else:
                print("Invalid choice!\n")
        except ODataError as odata_error:
            print("Error:")
            if odata_error.error:
                print(odata_error.error.code, odata_error.error.message)


async def greet_user(graph: Graph):
    user = await graph.get_user()
    if user:
        print("Hello,", user.display_name)
        # For Work/school accounts, email is in mail property
        # Personal accounts, email is in userPrincipalName
        print("Email:", user.mail or user.user_principal_name, "\n")


async def display_access_token(graph: Graph):
    token = await graph.get_user_token()
    print("User token:", token, "\n")


async def list_inbox(graph: Graph):
    message_page = await graph.get_inbox()
    if message_page and message_page.value:
        # Output each message's details
        for message in message_page.value:
            print("Message:", message.subject)
            if message.from_ and message.from_.email_address:
                print("  From:", message.from_.email_address.name or "NONE")
            else:
                print("  From: NONE")
            print("  Status:", "Read" if message.is_read else "Unread")
            print("  Received:", message.received_date_time)

        # If @odata.nextLink is present
        more_available = message_page.odata_next_link is not None
        print("\nMore messages available?", more_available, "\n")


# Run main
asyncio.run(main())

graph.py

from configparser import SectionProxy
from azure.identity import DeviceCodeCredential
from msgraph import graph_service_client

from msgraph.generated.users.item.user_item_request_builder import (
    UserItemRequestBuilder,
)
from msgraph.generated.users.item.mail_folders.item.messages.messages_request_builder import (
    MessagesRequestBuilder,
)


class Graph:
    settings: SectionProxy
    device_code_credential: DeviceCodeCredential
    user_client: graph_service_client.GraphServiceClient

    def __init__(self, config: SectionProxy):
        self.settings = config
        client_id = self.settings["clientId"]
        tenant_id = self.settings["tenantId"]
        graph_scopes = self.settings["graphUserScopes"].split(" ")

        self.device_code_credential = DeviceCodeCredential(
            client_id, tenant_id=tenant_id
        )
        self.user_client = graph_service_client.GraphServiceClient(
            self.device_code_credential, graph_scopes
        )

    async def get_user_token(self):
        graph_scopes = self.settings["graphUserScopes"]
        access_token = self.device_code_credential.get_token(graph_scopes)
        return access_token.token

    async def get_user(self):
        # Only request specific properties using $select
        query_params = UserItemRequestBuilder.UserItemRequestBuilderGetQueryParameters(
            select=["displayName", "mail", "userPrincipalName"]
        )

        request_config = (
            UserItemRequestBuilder.UserItemRequestBuilderGetRequestConfiguration(
                query_parameters=query_params
            )
        )

        user = await self.user_client.me.get(request_configuration=request_config)
        return user

    async def get_inbox(self):
        query_params = MessagesRequestBuilder.MessagesRequestBuilderGetQueryParameters(
            # Only request specific properties
            select=["from", "isRead", "receivedDateTime", "subject"],
            # Get at most 25 results
            top=25,
            # Sort by received time, newest first
            orderby=["receivedDateTime DESC"],
        )
        request_config = (
            MessagesRequestBuilder.MessagesRequestBuilderGetRequestConfiguration(
                query_parameters=query_params
            )
        )

        messages = await self.user_client.me.mail_folders.by_mail_folder_id(
            "inbox"
        ).messages.get(request_configuration=request_config)
        return messages

When using the following script using a client secret:

import asyncio
import logging
from azure.identity import ClientSecretCredential
from msgraph.graph_service_client import GraphServiceClient
from msgraph.generated.models.message import Message
from msgraph.generated.models.item_body import ItemBody
from msgraph.generated.models.body_type import BodyType
from msgraph.generated.models.recipient import Recipient
from msgraph.generated.models.email_address import EmailAddress
from msgraph.generated.users.item.send_mail.send_mail_post_request_body import (
    SendMailPostRequestBody,
)


class NotificationManager:
    def __init__(self) -> None:
        self.client_id = "xxxxxx"
        self.tenant_id = "xxxxxx"
        self.client_secret = "xxxxxx"
        self.sender = "xxxxx"
        self.scopes = ["https://graph.microsoft.com/.default"]

    def get_client(self) -> GraphServiceClient:
        try:
            credential = ClientSecretCredential(
                tenant_id=self.tenant_id,
                client_id=self.client_id,
                client_secret=self.client_secret,
            )
            return GraphServiceClient(credential, self.scopes)
        except Exception as e:
            logging.error(f"Failed to get Graph client: {e}")
            raise

    async def send_email(
        self,
        graph_client: GraphServiceClient,
        recipient: str,
        subject: str,
        content: str,
    ) -> None:
        try:
            message = Message(
                subject=subject,
                importance="low",
                body=ItemBody(
                    content_type=BodyType.Html,
                    content=self.form_html_message(content),
                ),
                to_recipients=[
                    Recipient(email_address=EmailAddress(address=recipient))
                ],
            )

            request_body = SendMailPostRequestBody(
                message=message, save_to_sent_items=True
            )

            response = await graph_client.users.by_user_id(self.sender).send_mail.post(
                request_body
            )

            if response is None:
                print("Mail sent successfully.")
            else:
                print("Unexpected response:", response)
        except Exception as e:
            logging.error(
                f"Failed to send email from {self.sender} to {recipient}: {e}"
            )
            raise

    def form_html_message(self, content: str) -> str:
        return f"""
        <html>
            <head>
                <style>
                    body {{
                        font-family: Arial, sans-serif;
                        font-size: 14px;
                        color: #333;
                    }}
                    h1 {{
                        color: #0078d4;
                    }}
                </style>
            </head>
            <body>
                <h1>Test mail</h1>
                <p>{content}</p>
            </body>
        </html>
        """


async def main():
    manager = NotificationManager()
    client = manager.get_client()
    await manager.send_email(
        graph_client=client,
        recipient="******@securitydokudok.onmicrosoft.com",
        subject="Test Subject (App Auth)",
        content="This is a test email sent using Microsoft Graph SDK via client credentials.",
    )


if __name__ == "__main__":
    asyncio.run(main())

We also get the same 401.

Hope anyone can help us out!

//Andreas from DokuDok