AI Productivity Tools 2026: Microsoft 365 Copilot, Notion AI, and the AI-Powered Workplace

import requests
from msal import ConfidentialClientApplication

# Authenticate with Microsoft Entra ID (formerly Azure AD)
app = ConfidentialClientApplication(
    client_id="your-client-id",
    client_credential="your-client-secret",
    authority="https://login.microsoftonline.com/your-tenant-id"
)
token = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])

# Use Microsoft Graph API to access user messages for AI processing
headers = {"Authorization": f"Bearer {token['access_token']}"}

# Fetch the latest 20 emails from a user's inbox
response = requests.get(
    "https://graph.microsoft.com/v1.0/users/[email protected]/messages?"
    "$top=20&$select=subject,bodyPreview,receivedDateTime&$orderby=receivedDateTime DESC",
    headers=headers
)
messages = response.json()

# Process emails through an LLM for custom summarization
import openai

openai.api_type = "azure"
openai.api_base = "https://your-resource.openai.azure.com/"
openai.api_key = "your-key"

for msg in messages["value"][:5]:
    prompt = f"""Summarize this email in one sentence:
From: {msg['subject']}
Preview: {msg['bodyPreview'][:500]}
Summary:"""

    summary = openai.ChatCompletion.create(
        engine="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=100
    )
    print(f"{msg['subject']}: {summary.choices[0].message.content}")

import requests
import openai

NOTION_TOKEN = "secret_your_notion_token"
NOTION_DATABASE_ID = "your_database_id"
openai.api_key = "your-openai-key"

# Step 1: Generate content via LLM
prompt = "Write a project status update for Q2: our team completed the API migration, "
prompt += "resolved 23 production bugs, and improved p95 latency by 40%. "
prompt += "Make it a 3-paragraph executive summary."

response = openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": prompt}],
    max_tokens=500
)
generated_text = response.choices[0].message.content

# Step 2: Create a new page in Notion with the generated content
notion_headers = {
    "Authorization": f"Bearer {NOTION_TOKEN}",
    "Content-Type": "application/json",
    "Notion-Version": "2022-06-28"
}

page_data = {
    "parent": {"database_id": NOTION_DATABASE_ID},
    "properties": {
        "Title": {
            "title": [{"text": {"content": "Q2 Project Status Update"}}]
        },
        "Status": {
            "select": {"name": "Ready for Review"}
        }
    },
    "children": [
        {
            "object": "block",
            "type": "paragraph",
            "paragraph": {
                "rich_text": [{"type": "text", "text": {"content": generated_text}}]
            }
        }
    ]
}

response = requests.post(
    "https://api.notion.com/v1/pages",
    headers=notion_headers,
    json=page_data
)
print(f"Page created: {response.json()['url']}")

import google.generativeai as genai
import base64
from google.oauth2 import service_account
from googleapiclient.discovery import build

# Configure Gemini API
genai.configure(api_key="YOUR_GEMINI_API_KEY")

# Access Gmail via Google API
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']
creds = service_account.Credentials.from_service_account_file(
    'service-account.json', scopes=SCOPES)
service = build('gmail', 'v1', credentials=creds)

# Fetch unread emails
results = service.users().messages().list(
    userId='me', q='is:unread', maxResults=10).execute()
messages = results.get('messages', [])

# Process through Gemini for smart triage
model = genai.GenerativeModel('gemini-3.1-pro')
triage_prompt = """Classify each email below into one of:
- ACTION_REQUIRED: needs a reply or task
- FOR_INFO: read and file
- TRASH: spam or irrelevant

Emails:
"""

for msg_id in messages[:10]:
    msg = service.users().messages().get(userId='me', id=msg_id['id']).execute()
    headers = {h['name']: h['value'] for h in msg['payload']['headers']}
    snippet = msg.get('snippet', '')
    triage_prompt += f"- From: {headers.get('From', '')}\n  Subject: {headers.get('Subject', '')}\n  Snippet: {snippet[:200]}\n"

response = model.generate_content(triage_prompt)
print(response.text)

Summarize the following email thread into 3 bullet points:
- Key decisions made
- Action items and owners
- Open questions

Thread content:
[Paste email thread]

Rewrite the following text for a [executive / technical / general] audience.
Maintain all factual content but adjust:
- Vocabulary level
- Sentence length
- Technical depth

[Paste document]

Convert these raw meeting notes into structured meeting minutes with:
- Date and attendees
- Discussion summary (3-4 sentences)
- Key decisions
- Action items (with owners)

Raw notes:
[Paste notes]

Analyze this data and identify:
1. Top 3 trends
2. Any anomalies or outliers
3. Recommended actions

Data: [Paste data range]

import smtplib
from email.mime.text import MIMEText
from typing import List, Dict
import openai

class EmailAgent:
    def __init__(self, api_key: str, smtp_config: dict):
        openai.api_key = api_key
        self.smtp = smtp_config

    def process_inbox(self, emails: List[Dict]) -> List[Dict]:
        results = []
        for email in emails:
            classification = self.classify_email(email)
            if classification == "ACTION_REQUIRED":
                draft = self.draft_reply(email)
                results.append({
                    "email_id": email["id"],
                    "action": "drafted",
                    "draft": draft,
                    "priority": self.compute_priority(email)
                })
            elif classification == "MEETING_REQUEST":
                slots = self.suggest_meeting_times(email)
                results.append({
                    "email_id": email["id"],
                    "action": "meeting_proposed",
                    "slots": slots
                })
            else:
                results.append({
                    "email_id": email["id"],
                    "action": "archived"
                })
        return results

    def classify_email(self, email: Dict) -> str:
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": "Classify the email. Return one: ACTION_REQUIRED, "
                           "FOR_INFO, MEETING_REQUEST, TRASH."
            }, {
                "role": "user",
                "content": f"Subject: {email['subject']}\nBody: {email['body'][:500]}"
            }],
            temperature=0.1
        )
        return response.choices[0].message.content.strip()

    def draft_reply(self, email: Dict) -> str:
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": "Draft a professional, concise email reply. "
                           "Use the same language as the original email."
            }, {
                "role": "user",
                "content": f"Original email:\nSubject: {email['subject']}\n"
                           f"Body: {email['body']}\n\nDraft reply:"
            }],
            temperature=0.3
        )
        return response.choices[0].message.content.strip()

    def compute_priority(self, email: Dict) -> str:
        keywords_high = {"urgent", "asap", "deadline", "critical", "blocker"}
        body_lower = email.get("body", "").lower()
        if any(kw in body_lower for kw in keywords_high):
            return "high"
        if email.get("from") in self.vip_senders:
            return "high"
        return "normal"

agent = EmailAgent(
    api_key="sk-...",
    smtp_config={"host": "smtp.example.com", "port": 587}
)

# Fetch emails via Graph API or IMAP, then process
emails = fetch_unread_emails()
results = agent.process_inbox(emails)
for r in results:
    print(f"{r['email_id']}: {r['action']} — {r.get('priority', 'normal')}")

from datetime import datetime, timedelta
import openai
import requests

class MeetingLifecycleAgent:
    def __init__(self, calendar_api_url: str, openai_key: str):
        self.cal_url = calendar_api_url
        openai.api_key = openai_key

    def before_meeting(self, meeting_id: str) -> dict:
        """Generate pre-read brief from calendar event context."""
        event = self.get_event(meeting_id)
        recent_emails = self.get_related_emails(event["title"], event["attendees"])
        context = f"Meeting: {event['title']}\nAttendees: {', '.join(event['attendees'])}"

        brief = openai.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": "Generate a one-paragraph pre-read brief summarizing "
                           "recent context and suggesting discussion points."
            }, {
                "role": "user",
                "content": context + "\n\nRecent emails:\n" + "\n".join(recent_emails[:3])
            }],
            max_tokens=300
        )
        return {"brief": brief.choices[0].message.content}

    def during_meeting(self, transcript: str) -> dict:
        """Real-time transcription analysis for action items."""
        analysis = openai.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": "Extract from this transcript:\n1. Key decisions\n2. Action items with owners\n3. Deadlines mentioned\n4. Open questions"
            }, {
                "role": "user",
                "content": transcript
            }],
            temperature=0.1
        )
        return {"analysis": analysis.choices[0].message.content}

    def after_meeting(self, analysis: dict, calendar_id: str) -> None:
        """Create follow-up tasks and update calendar."""
        tasks = self.parse_action_items(analysis["analysis"])
        for task in tasks:
            self.create_task(task["owner"], task["description"], task.get("deadline"))

    def get_event(self, meeting_id):
        # Fetch event from calendar API
        response = requests.get(f"{self.cal_url}/events/{meeting_id}")
        return response.json()

    def get_related_emails(self, title, attendees, limit=5):
        # Fetch emails related to this meeting
        query = f"subject:'{title}' OR from:{' OR from:'.join(attendees)}"
        response = requests.get(f"https://graph.microsoft.com/v1.0/me/messages?$search={query}")
        return [msg['bodyPreview'] for msg in response.json().get('value', [])[:limit]]

    def parse_action_items(self, analysis_text):
        # Parse structured action items from analysis
        items = []
        for line in analysis_text.split('\n'):
            if line.strip().startswith('-') or line.strip().startswith('*'):
                items.append({"description": line.strip().lstrip('- *'), "owner": "auto-assigned"})
        return items

    def create_task(self, owner, description, deadline=None):
        # Create task via API
        task_data = {"title": description, "assignee": owner, "due": deadline}
        requests.post(f"{self.cal_url}/tasks", json=task_data)

# Pseudocode for a Zapier Central agent that handles customer onboarding
agent_config = {
    "trigger": "new_customer_in_crm",
    "steps": [
        {
            "action": "ai_decide",
            "prompt": "Analyze the customer's industry and company size. "
                      "Determine if they need: 'technical_onboarding', "
                      "'standard_onboarding', or 'enterprise_onboarding'."
        },
        {
            "action": "conditional",
            "condition": "{output == 'technical_onboarding'}",
            "then": [
                {"action": "create_ticket_in_linear", "assignee": "tech-team"},
                {"action": "send_slack_message", "channel": "#tech-onboarding"},
                {"action": "send_email", "template": "tech-welcome"}
            ]
        },
        {
            "action": "ai_generate",
            "prompt": "Write a personalized onboarding timeline "
                      "based on the customer's selected plan and industry.",
            "output_as": "timeline"
        },
        {
            "action": "create_in_notion",
            "database": "Onboarding Projects",
            "content": "{timeline}"
        }
    ]
}

# Check data protection before sending to AI API
def should_send_to_ai(content: str, classification: str) -> bool:
    """Only send non-sensitive content through external AI APIs."""
    SENSITIVE_CLASSIFICATIONS = {"confidential", "restricted", "pii"}
    if classification.lower() in SENSITIVE_CLASSIFICATIONS:
        return False
    if re.search(r'\b\d{3}-\d{2}-\d{4}\b', content):  # SSN pattern
        return False
    return True

from flask import Flask, request, jsonify
import openai
import re
import logging

app = Flask(__name__)

# Governance rules engine
SENSITIVE_PATTERNS = {
    "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
    "credit_card": r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b',
    "email": r'\b[\w\.-]+@[\w\.-]+\.\w+\b',
    "phone": r'\b\+?\d[\d\s()-]{7,}\d\b'
}

ALLOWED_MODELS = {
    "gpt-4o": {"max_requests_per_hour": 1000, "classification": "internal"},
    "gpt-4o-mini": {"max_requests_per_hour": 5000, "classification": "internal"},
    "claude-sonnet-4-6": {"max_requests_per_hour": 500, "classification": "confidential"},
}

@app.route('/v1/chat/completions', methods=['POST'])
def governed_chat():
    data = request.json
    user = request.headers.get('X-User-Id')
    dept = request.headers.get('X-Department')

    # Check rate limits per user
    if is_rate_limited(user, data['model']):
        return jsonify({"error": "Rate limit exceeded"}), 429

    # Scan content for sensitive data
    content = str(data.get('messages', ''))
    for pattern_name, pattern in SENSITIVE_PATTERNS.items():
        if re.search(pattern, content):
            logging.warning(f"Blocked request from {user}: {pattern_name} detected")
            return jsonify({
                "error": f"Request blocked: {pattern_name} detected in prompt",
                "allowed": False
            }), 400

    # Route through enterprise endpoint with data isolation
    response = openai.chat.completions.create(
        model=data['model'],
        messages=data['messages'],
        extra_headers={"OpenAI-Organization-Isolation": "true"}
    )

    # Log for audit
    log_audit_event(user, dept, data['model'], len(content), response.usage)

    return jsonify(response.to_dict())

class BudgetController:
    def __init__(self, monthly_budget: float):
        self.monthly_budget = monthly_budget
        self.usage = {}  # dept -> cost

    def can_process(self, dept: str, estimated_cost: float) -> bool:
        dept_usage = self.usage.get(dept, 0)
        dept_budget = self.monthly_budget * 0.3  # 30% per dept max

        if dept_usage + estimated_cost > dept_budget:
            return False
        return True

    def track_usage(self, dept: str, tokens_in: int, tokens_out: int, model: str):
        cost = self.calculate_cost(tokens_in, tokens_out, model)
        self.usage[dept] = self.usage.get(dept, 0) + cost
        if self.usage[dept] > self.monthly_budget * 0.25:
            self.send_alert(dept, self.usage[dept])

    def calculate_cost(self, tokens_in, tokens_out, model):
        rates = {"gpt-4o": (2.50, 10.00), "gpt-4o-mini": (0.15, 0.60)}
        in_rate, out_rate = rates.get(model, (2.50, 10.00))
        return (tokens_in / 1_000_000 * in_rate) + (tokens_out / 1_000_000 * out_rate)