前置条件：
- 有一个365邮箱，配置好许可证
- 在azure portal里有Microsoft Entra ID ，注册相关应用时graph API赋权
- 应用的应用程序(客户端) ID，目录(租户) ID，客户端的密码，邮箱的id，名称
1.收件：
import os
import json
import requests
from .auth2 import get_access_tokendef load_token_from_file():TOKEN_FILE = 'token_cache.json'if os.path.exists(TOKEN_FILE):with open(TOKEN_FILE, 'r') as f:data = json.load(f)return data.get('access_token', None)return Nonedef fetch_unread_emails():try:access_token = get_access_token()if not access_token:print('无法加载访问令牌。')returnheaders = {'Authorization': f'Bearer {access_token}','Content-Type': 'application/json'}user_id = '-------'url = f'https://graph.microsoft.com/v1.0/users/{user_id}/mailfolders/inbox/messages?$filter=isRead eq false'response = requests.get(url, headers=headers)print(f"Response Status Code: {response.status_code}")print(f"Response Content: {response.content}")if response.status_code != 200:print(f"Error fetching emails: {response.content}")return
###代码仅为示例，如想了解更多收发邮件功能，实现赋能gpt的自动收发，处理，function calling，add wx： MTMwMTE4MjY1OTI= （base64）emails = response.json().get('value', [])for email in emails:print(f"Email ID: {email['id']}")print(f"From: {email['from']['emailAddress']['address']}")print(f"Subject: {email['subject']}")print(f"Content: {email['body']['content']}")print(f"Received Time: {email['receivedDateTime']}")print("-" * 40)except Exception as ex:print("Error fetching unread emails: ", ex)# Test fetching unread emails
if __name__ == "__main__":fetch_unread_emails()2.发件：
from flask import Flask, request, jsonify
import requests
import base64
import msal
import osapp = Flask(__name__)# Configuration variables
CLIENT_ID = 'your_client_id'
CLIENT_SECRET = 'your_client_secret'
TENANT_ID = 'your_tenant_id'
SCOPES = ['https://graph.microsoft.com/.default']
CACHE_FILE = 'token_cache.bin'
URL = 'https://graph.microsoft.com/v1.0/me/sendMail'def load_cache():cache = msal.SerializableTokenCache()if os.path.exists(CACHE_FILE):cache.deserialize(open(CACHE_FILE, "r").read())return cachedef save_cache(cache):if cache.has_state_changed:with open(CACHE_FILE, "w") as f:f.write(cache.serialize())def get_access_token():cache = load_cache()app = msal.ConfidentialClientApplication(CLIENT_ID,authority=f"https://login.microsoftonline.com/{TENANT_ID}",client_credential=CLIENT_SECRET,token_cache=cache)result = app.acquire_token_silent(SCOPES, account=None)if not result:result = app.acquire_token_for_client(scopes=SCOPES)if "access_token" in result:save_cache(cache)return result['access_token']else:raise ValueError(f"Failed to obtain access token: {result.get('error_description')}")def create_attachment(file_path, content_id):with open(file_path, "rb") as f:content_bytes = base64.b64encode(f.read()).decode('utf-8')return {"@odata.type": "#microsoft.graph.fileAttachment","name": os.path.basename(file_path),"contentBytes": content_bytes,"contentId": content_id}@app.route('/send-email', methods=['POST'])
def send_email_api():data = request.jsontry:token = get_access_token()sender_email = data['sender']recipients_emails = data['recipients']subject = data['subject']body = data['body']attachments = [create_attachment(file_path, content_id) for file_path, content_id in data.get('attachments', [])]email_data = {"message": {"subject": subject,"body": {"contentType": "HTML","content": body},"toRecipients": [{"emailAddress": {"address": recipient}} for recipient in recipients_emails],"from": {"emailAddress": {"address": sender_email}},"attachments": attachments}}headers = {"Authorization": f"Bearer {token}","Content-Type": "application/json"}response = requests.post(URL, headers=headers, json=email_data)if response.status_code == 202:return jsonify({"message": "Email sent successfully"}), 200else:return jsonify({"error": f"Failed to send email: {response.status_code}"}), 500except Exception as e:return jsonify({"error": str(e)}), 500if __name__ == "__main__":app.run(debug=True, port=5000)
