Failing to read outlook messages using msgraph and msal in Python

Question

Can anyone please advise where I'm going wrong? I've been trying to access my emails programatically using Python's Requests library and MSAL querying the f"users/{self.user_id}/messages" endpoint but whatever I do I keep getting 500 API response

When I log into graph explorer and copy the bearer token from there and plug it into the code it works so it will be to do with the token creation but I don't know where the issue is

Here is part of the code

import os
import msal
import requests
from loguru import logger
from typing import List, Dict
from dataclasses import dataclass
from datetime import datetime, timedelta
from jose import jwt
@dataclass
class Email:
    id: str
    sender: str
    subject: str
    body: str
    received_time: datetime
    attachments: List[Dict]
class EmailClient:
    def __init__(self):
        # MSAL configuration
        self.client_id = os.getenv("AZURE_CLIENT_ID")
        self.client_secret = os.getenv("AZURE_CLIENT_SECRET")
        self.tenant_id = os.getenv("AZURE_TENANT_ID")
        self.user_id = os.getenv("EMAIL_USERNAME")
        self.authority = f"https://login.microsoftonline.com/{self.tenant_id}"
        self.scopes = ["https://graph.microsoft.com/.default"]
        self.folder_name = os.getenv("EMAIL_FOLDER", "Inbox")
        self.graph_api_endpoint = "https://graph.microsoft.com/v1.0"
        if not all([self.client_id, self.client_secret, self.tenant_id, self.user_id]):
            raise ValueError("Missing required environment variables: AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID, or USER_ID")
        
        try:
            # Initialize MSAL app
            self.app = msal.ConfidentialClientApplication(
                client_id=self.client_id,
                client_credential=self.client_secret,
                authority=self.authority
            )
            # Get token for Microsoft Graph API
            result = self.app.acquire_token_for_client(scopes=self.scopes)
            if "access_token" not in result:
                raise ValueError(f"Failed to obtain access token: {result.get('error_description', 'Unknown error')}")
            self.token = result["access_token"]
            logger.info(f"Token acquired successfully")
            # Decode the token to inspect its contents
            try:
                decoded_token = jwt.get_unverified_claims(self.token)
                logger.info(f"Token scopes: {decoded_token.get('roles')}")
            except Exception as e:
                logger.warning(f"Could not decode token: {str(e)}")
            logger.info("Successfully authenticated with Microsoft Graph API using MSAL")
        
        except Exception as e:
            logger.error(f"Failed to initialize email client: {str(e)}")
            raise
    def _make_request(self, method, endpoint, params=None, data=None):
        """Make HTTP request to Microsoft Graph API"""
        url = f"{self.graph_api_endpoint}/{endpoint}"
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json",
            "Prefer": "outlook.body-content-type=\"html\""
        }
        response = requests.request(method=method, url=url, headers=headers, params=params, json=data)
        logger.info(f"Request URL: {url}")
        logger.info(f"Request headers: {headers}")
        logger.info(f"Request parameters: {params}")
        logger.info(f"Request data: {data}")
        logger.info(f"Response: {response.text}")
        # Check response status and log or handle errors
        try:
            response.raise_for_status()
        except requests.HTTPError as e:
            logger.error(f"API request failed: {response.status_code} - {response.text}")
            return None
        # Attempt to parse JSON response
        try:
            return response.json()
        except ValueError as e:
            logger.error(f"Failed to decode JSON from response: {e}")
            logger.error(f"Response content: '{response.text}'")
            return None
    def fetch_new_emails(self) -> List[Email]:
        """Fetch new emails from the specified folder"""
        try:
            # Calculate time 24 hours ago in ISO format
            cutoff = (datetime.now() - timedelta(days=1)).isoformat() + 'Z'
            logger.info(f"Fetching emails for user {self.user_id} since {cutoff}")
            # Build the query parameters
            params = {
                "$top": 50,
                "$select": "id,subject,from,receivedDateTime,hasAttachments",
                "$orderby": "receivedDateTime desc"
            }
            # Get messages
            endpoint = f"users/{self.user_id}/messages"
            messages = self._make_request("GET", endpoint, params=params)
            if not messages or "value" not in messages:
                logger.warning(f"No messages found for user {self.user_id}")
                return []
            logger.info(f"Found {len(messages['value'])} messages for user {self.user_id}")
            emails = []
            for msg in messages["value"]:
                try:
                    # Get full message details
                    msg_id = msg["id"]
                    full_msg = self._make_request("GET", f"users/{self.user_id}/messages/{msg_id}")
                    attachments = []
                    if msg.get("hasAttachments", False):
                        # Get attachments
                        atts = self._make_request("GET", f"users/{self.user_id}/messages/{msg_id}/attachments")
                        if atts and "value" in atts:
                            attachments = [
                                {
                                    "filename": att["name"],
                                    "content": att.get("contentBytes", ""),
                                    "content_type": att.get("contentType", "")
                                }
                                for att in atts["value"]
                            ]
                    
                    # Use the 'from' field from the initial message list if it's not in the full message
                    sender = msg.get("from", {}).get("emailAddress", {}).get("address", "unknown")
                    # Get the body content from the full message
                    body_content = full_msg.get("body", {}).get("content", "")
                    emails.append(Email(
                        id=msg_id,
                        sender=sender,
                        subject=msg["subject"],
                        body=body_content,
                        received_time=datetime.fromisoformat(msg["receivedDateTime"].replace("Z", "+00:00")),
                        attachments=attachments
                    ))
                except Exception as e:
                    logger.error(f"Error processing message {msg.get('id', 'unknown')}: {str(e)}")
                    # Continue with the next message
            return emails
        except Exception as e:
            logger.error(f"Error fetching emails: {str(e)}")
            return []

Here is the error I get when I run my pytest

Any help/advice would be be much appreciated

Answer 1

Hello @Robert Kovacs,

I understand that you are unable to read outlook messages using Microsoft Graph and MSAL in Python make sure to grant Mail.Read application type API permission to the Microsoft Entra ID application:

And I used the below code to read the messages:

import os

import msal

import requests

from loguru import logger

from typing import List, Dict

from dataclasses import dataclass

from datetime import datetime, timedelta

from jose import jwt

 

 

@dataclass

class Email:

    sender: str

    subject: str

    received_time: datetime

 

 

class EmailClient:

    def __init__(self, client_id: str, client_secret: str, tenant_id: str, user_id: str, folder_name: str = "Inbox"):

        # Direct assignment of config variables

        self.client_id = client_id

        self.client_secret = client_secret

        self.tenant_id = tenant_id

        self.user_id = user_id

        self.folder_name = folder_name

        self.authority = f"https://login.microsoftonline.com/{self.tenant_id}"

        self.scopes = ["https://graph.microsoft.com/.default"]

        self.graph_api_endpoint = "https://graph.microsoft.com/v1.0"

 

        try:

            # Initialize MSAL app

            self.app = msal.ConfidentialClientApplication(

                client_id=self.client_id,

                client_credential=self.client_secret,

                authority=self.authority

            )

            # Get token for Microsoft Graph API

            result = self.app.acquire_token_for_client(scopes=self.scopes)

            if "access_token" not in result:

                raise ValueError(f"Failed to obtain access token: {result.get('error_description', 'Unknown error')}")

            self.token = result["access_token"]

            logger.info(f"Token acquired successfully")

 

            # Decode token for inspection

            try:

                decoded_token = jwt.get_unverified_claims(self.token)

                logger.info(f"Token scopes: {decoded_token.get('roles')}")

            except Exception as e:

                logger.warning(f"Could not decode token: {str(e)}")

 

            logger.info("Successfully authenticated with Microsoft Graph API using MSAL")

        except Exception as e:

            logger.error(f"Failed to initialize email client: {str(e)}")

            raise

 

    def _make_request(self, method, endpoint, params=None, data=None):

        """Make HTTP request to Microsoft Graph API"""

        url = f"{self.graph_api_endpoint}/{endpoint}"

        headers = {

            "Authorization": f"Bearer {self.token}",

            "Content-Type": "application/json",

            "Prefer": "outlook.body-content-type=\"html\""

        }

        response = requests.request(method=method, url=url, headers=headers, params=params, json=data)

        logger.info(f"Request URL: {url}")

        logger.info(f"Request headers: {headers}")

        logger.info(f"Request parameters: {params}")

        logger.info(f"Request data: {data}")

        logger.info(f"Response: {response.text}")

        # Check response status and log or handle errors

        try:

            response.raise_for_status()

        except requests.HTTPError as e:

            logger.error(f"API request failed: {response.status_code} - {response.text}")

            return None

        # Attempt to parse JSON response

        try:

            return response.json()

        except ValueError as e:

            logger.error(f"Failed to decode JSON from response: {e}")

            logger.error(f"Response content: '{response.text}'")

            return None

 

    def fetch_new_emails(self) -> List[Email]:

        """Fetch new emails from the specified folder"""

        try:

            # Calculate time 24 hours ago in ISO format

            cutoff = (datetime.now() - timedelta(days=1)).isoformat() + 'Z'

            logger.info(f"Fetching emails for user {self.user_id} since {cutoff}")

            # Build the query parameters

            params = {

                "$top": 50,

                "$select": "id,subject,from,receivedDateTime",

                "$orderby": "receivedDateTime desc"

            }

            # Get messages

            endpoint = f"users/{self.user_id}/messages"

            messages = self._make_request("GET", endpoint, params=params)

            if not messages or "value" not in messages:

                logger.warning(f"No messages found for user {self.user_id}")

                return []

            logger.info(f"Found {len(messages['value'])} messages for user {self.user_id}")

            emails = []

            for msg in messages["value"]:

                try:

                    # Get sender and subject details from the initial message list

                    sender = msg.get("from", {}).get("emailAddress", {}).get("address", "unknown")

                    subject = msg.get("subject", "No subject")

                    received_time = datetime.fromisoformat(msg["receivedDateTime"].replace("Z", "+00:00"))

                    # Append simplified email details

                    emails.append(Email(

                        sender=sender,

                        subject=subject,

                        received_time=received_time

                    ))

                except Exception as e:

                    logger.error(f"Error processing message {msg.get('id', 'unknown')}: {str(e)}")

                    # Continue with the next message

            return emails

        except Exception as e:

            logger.error(f"Error fetching emails: {str(e)}")

            return []

 

 

# Example Usage:

if __name__ == "__main__":

    # Replace these values with actual credentials

    client = EmailClient(

        client_id="ClientID",

        client_secret="Secret",

        tenant_id="TenantID",

        user_id="UserID",

        folder_name="Inbox"  

    )

 

    emails = client.fetch_new_emails()

    for email in emails:

        # Printing in the requested format

        print(f"From: {email.sender}, Subject: {email.subject}, Received: {email.received_time}")

I am able to read messages successfully:

Reference:

List messages - Microsoft Graph v1.0 | Microsoft

Hope this helps!

---

If this answer was helpful, please click "Accept the answer" and mark Yes, as this can be beneficial to other community members.

If you have any other questions or still running into more issues, let me know in the "comments" and I would be happy to help you.