Great question. In Prefect 2.x there isn’t a built‑in Outlook/Office 365 connector, so the usual approach is to call Microsoft Graph from a Prefect flow, authenticate with Azure AD, then download the email attachment(s). Below is a working example using: - Microsoft Graph + MSAL (client credentials / app-only) - Prefect Blocks to store your secrets - A flow that finds the most recent message with attachments and downloads them High-level steps 1) Create an Azure AD app registration and grant Graph permissions - If you can use app-only auth, assign Application permissions like Mail.Read (admin consent required). To limit access to specific mailboxes, use an Application Access Policy. - If you must act as a user, use delegated permissions (e.g., Mail.Read) and an interactive device code/auth flow instead. Docs: - Microsoft Graph overview - Mail permissions (Graph) - Application access policies (limit app to specific mailboxes) 2) Store your credentials in Prefect 2.x as Secret blocks - client_id, client_secret, tenant_id (and optionally a mailbox email) 3) Run a Prefect flow that: - Gets an access token with MSAL - Finds the most recent message with attachments - Downloads the attachments to disk (or push to your storage) Example: Prefect 2.x flow using Microsoft Graph Dependencies: - prefect>=2 - msal - requests How to create Secret blocks (do this once, ideally in a notebook or REPL; or use the UI):

from prefect.blocks.system import Secret

Secret(value="YOUR_TENANT_ID").save("ms-tenant-id", overwrite=True)
Secret(value="YOUR_CLIENT_ID").save("ms-client-id", overwrite=True)
Secret(value="YOUR_CLIENT_SECRET").save("ms-client-secret", overwrite=True)

Flow code: ``` import os import base64 from pathlib import Path import requests import msal from prefect import flow, task from prefect.blocks.system import Secret GRAPH_SCOPE = ["https://graph.microsoft.com/.default"] GRAPH_BASE = "https://graph.microsoft.com/v1.0" @task(retries=3, retry_delay_seconds=30) def get_graph_token(tenant_id: str, client_id: str, client_secret: str) -> str: authority = f"https://login.microsoftonline.com/{tenant_id}" app = msal.ConfidentialClientApplication( client_id=client_id, authority=authority, client_credential=client_secret, ) result = app.acquire_token_for_client(scopes=GRAPH_SCOPE) if "access_token" not in result: raise RuntimeError(f"Failed to get token: {result.get('error')}: {result.get('error_description')}") return result["access_token"] @task def find_latest_message_with_attachments(access_token: str, mailbox: str) -> str: url = f"{GRAPH_BASE}/users/{mailbox}/messages" params = { "$filter": "hasAttachments eq true", "$orderby": "receivedDateTime desc", "$top": "1", "$select": "id,subject,receivedDateTime" } headers = {"Authorization": f"Bearer {access_token}"} r = requests.get(url, headers=headers, params=params, timeout=30) r.raise_for_status() items = r.json().get("value", []) if not items: raise RuntimeError("No messages with attachments found.") return items[0]["id"] @task def download_attachments(access_token: str, mailbox: str, message_id: str, download_dir: str = "./downloads") -> list[str]: Path(download_dir).mkdir(parents=True, exist_ok=True) url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}/attachments" params = {"$filter": "isInline eq false"} headers = {"Authorization": f"Bearer {access_token}"} r = requests.get(url, headers=headers, params=params, timeout=30) r.raise_for_status() saved = [] for att in r.json().get("value", []): # Only FileAttachment has contentBytes; ItemAttachment will not

content_b64 = att.get("contentBytes") name = att.get("name", "attachment.bin") if content_b64: path = os.path.join(download_dir, name) with open(path, "wb") as f: f.write(base64.b64decode(content_b64)) saved.append(path) if not saved: raise RuntimeError("Message has no downloadable file attachments.") return saved @flow def download_outlook_attachments_flow(mailbox: str, download_dir: str = "./downloads") -> list[str]: tenant_id = Secret.load("ms-tenant-id").get() client_id = Secret.load("ms-client-id").get() client_secret = Secret.load("ms-client-secret").get() token = get_graph_token(tenant_id, client_id, client_secret) message_id = find_latest_message_with_attachments(token, mailbox) files = download_attachments(token, mailbox, message_id, download_dir) return files if name == "__main__": # Example mailbox: a user or shared mailbox address download_outlook_attachments_flow(mailbox="shared-mailbox@yourtenant.com", download_dir="./downloads") ``` Notes and alternatives - Filtering/selection: Update the Graph query to match your use case (e.g., subject contains, from address, date). See: - List messages (Graph) - List attachments (Graph) - Delegated auth: If you must act on behalf of a user, use MSAL’s device code or authorization code flows and store/refresh the token. See MSAL for Python. - Library wrappers: You can also use the O365 Python library, which wraps Graph and simplifies mailbox/attachment operations: python-o365. This can be helpful if you prefer a higher-level API. - IMAP/basic auth: Often disabled in Microsoft 365 tenants and not recommended. Would you share a few details so I can tailor this to your setup? - Is the mailbox a user mailbox or a shared mailbox? - Can you use app-only authentication (recommended) or do you need delegated user auth? - Do you want the file saved locally, or uploaded to S3/GCS/Azure Blob? - Any specific filters (subject/sender/file type) you want to apply?