"""Microsoft Graph API tools using per-user OAuth tokens.
Provides OneDrive, Outlook, and Calendar operations via the Microsoft
Graph API. Requires the user to have connected their Microsoft account
via the OAuth flow.
"""
from __future__ import annotations
import json
import logging
import base64
from typing import Any, TYPE_CHECKING
from urllib.parse import urlparse
import aiohttp
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
GRAPH_API = "https://graph.microsoft.com/v1.0"
async def _ms_request(
method: str,
path: str,
token: str,
*,
params: dict[str, Any] | None = None,
json_body: dict[str, Any] | None = None,
data: bytes | None = None,
extra_headers: dict[str, str] | None = None,
) -> dict[str, Any] | list[Any] | str:
headers = {"Authorization": f"Bearer {token}"}
if extra_headers:
headers.update(extra_headers)
if path.startswith("http"):
parsed = urlparse(path)
if (parsed.hostname or "").lower() != "graph.microsoft.com":
return {
"error": "Only graph.microsoft.com requests are allowed",
"detail": parsed.hostname or "",
}
url = path
else:
url = f"{GRAPH_API}{path}"
async with aiohttp.ClientSession() as session:
kwargs: dict[str, Any] = {"headers": headers, "params": params}
if json_body is not None:
kwargs["json"] = json_body
if data is not None:
kwargs["data"] = data
async with session.request(method, url, **kwargs) as resp:
if resp.status == 204:
return {"status": "success", "code": 204}
body = await resp.text()
if resp.status >= 400:
return {"error": f"Microsoft Graph error ({resp.status})", "detail": body[:2000]}
try:
return json.loads(body)
except json.JSONDecodeError:
return body[:4000]
async def _get_token(ctx: ToolContext | None) -> str:
if ctx is None or ctx.redis is None:
raise RuntimeError("Context or Redis not available")
from oauth_manager import require_oauth_token
return await require_oauth_token(ctx, "microsoft")
# ---------------------------------------------------------------------------
# OneDrive handlers
# ---------------------------------------------------------------------------
[docs]
async def microsoft_onedrive_list(
path: str = "",
search: str = "",
limit: int = 20,
ctx: ToolContext | None = None,
) -> str:
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
if search:
if "\n" in search or "\r" in search:
return json.dumps({"error": "Invalid search string"})
q_esc = search.replace("\\", "\\\\").replace("'", "''")
endpoint = f"/me/drive/root/search(q='{q_esc}')"
params = {"$top": min(limit, 200)}
elif path:
if "\n" in path or "\r" in path or ".." in path:
return json.dumps({"error": "Invalid OneDrive path"})
endpoint = f"/me/drive/root:/{path}:/children"
params = {"$top": min(limit, 200)}
else:
endpoint = "/me/drive/root/children"
params = {"$top": min(limit, 200)}
data = await _ms_request("GET", endpoint, token, params=params)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
items = []
for item in data.get("value", []) if isinstance(data, dict) else []:
entry: dict[str, Any] = {
"id": item.get("id"),
"name": item.get("name"),
"size": item.get("size"),
"lastModifiedDateTime": item.get("lastModifiedDateTime"),
"webUrl": item.get("webUrl"),
}
if "folder" in item:
entry["type"] = "folder"
entry["childCount"] = item["folder"].get("childCount")
else:
entry["type"] = "file"
entry["mimeType"] = item.get("file", {}).get("mimeType")
items.append(entry)
return json.dumps({"count": len(items), "items": items})
[docs]
async def microsoft_onedrive_read(
item_id: str = "",
path: str = "",
ctx: ToolContext | None = None,
) -> str:
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
if item_id:
if "/" in item_id or ".." in item_id or "\n" in item_id:
return json.dumps({"error": "Invalid item_id"})
meta_endpoint = f"/me/drive/items/{item_id}"
elif path:
if "\n" in path or "\r" in path or ".." in path:
return json.dumps({"error": "Invalid OneDrive path"})
meta_endpoint = f"/me/drive/root:/{path}"
else:
return json.dumps({"error": "Provide either item_id or path"})
meta = await _ms_request("GET", meta_endpoint, token)
if isinstance(meta, dict) and "error" in meta:
return json.dumps(meta)
download_url = meta.get("@microsoft.graph.downloadUrl", "") if isinstance(meta, dict) else ""
content = ""
if download_url:
async with aiohttp.ClientSession() as session:
async with session.get(download_url) as resp:
if resp.status == 200:
raw = await resp.read()
try:
content = raw.decode("utf-8", errors="replace")[:16000]
except Exception:
content = "(binary file)"
return json.dumps({
"metadata": {
"id": meta.get("id") if isinstance(meta, dict) else None,
"name": meta.get("name") if isinstance(meta, dict) else None,
"size": meta.get("size") if isinstance(meta, dict) else None,
"mimeType": meta.get("file", {}).get("mimeType") if isinstance(meta, dict) else None,
"webUrl": meta.get("webUrl") if isinstance(meta, dict) else None,
},
"content": content,
})
[docs]
async def microsoft_onedrive_upload(
name: str,
content: str,
folder_path: str = "",
ctx: ToolContext | None = None,
) -> str:
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
if "\n" in name or "\r" in name or ".." in name:
return json.dumps({"error": "Invalid file name"})
if folder_path:
if "\n" in folder_path or "\r" in folder_path or ".." in folder_path:
return json.dumps({"error": "Invalid folder_path"})
endpoint = f"/me/drive/root:/{folder_path}/{name}:/content"
else:
endpoint = f"/me/drive/root:/{name}:/content"
data = await _ms_request(
"PUT", endpoint, token,
data=content.encode(),
extra_headers={"Content-Type": "application/octet-stream"},
)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
return json.dumps({
"status": "uploaded",
"id": data.get("id") if isinstance(data, dict) else None,
"name": data.get("name") if isinstance(data, dict) else name,
"webUrl": data.get("webUrl") if isinstance(data, dict) else None,
"size": data.get("size") if isinstance(data, dict) else len(content),
})
# ---------------------------------------------------------------------------
# Outlook handlers
# ---------------------------------------------------------------------------
[docs]
async def microsoft_outlook_list(
folder: str = "inbox",
search: str = "",
limit: int = 15,
skip: int = 0,
ctx: ToolContext | None = None,
) -> str:
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
endpoint = f"/me/mailFolders/{folder}/messages"
params: dict[str, Any] = {
"$top": min(limit, 50),
"$skip": skip,
"$select": "id,subject,from,toRecipients,receivedDateTime,bodyPreview,isRead,hasAttachments",
"$orderby": "receivedDateTime desc",
}
if search:
params["$search"] = f'"{search}"'
data = await _ms_request("GET", endpoint, token, params=params)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
messages = []
for m in data.get("value", []) if isinstance(data, dict) else []:
messages.append({
"id": m.get("id"),
"subject": m.get("subject"),
"from": m.get("from", {}).get("emailAddress", {}).get("address"),
"to": [r.get("emailAddress", {}).get("address") for r in m.get("toRecipients", [])],
"receivedDateTime": m.get("receivedDateTime"),
"preview": m.get("bodyPreview", "")[:200],
"isRead": m.get("isRead"),
"hasAttachments": m.get("hasAttachments"),
})
return json.dumps({"count": len(messages), "messages": messages})
[docs]
async def microsoft_outlook_read(
message_id: str,
ctx: ToolContext | None = None,
) -> str:
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
data = await _ms_request("GET", f"/me/messages/{message_id}", token)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
body_content = data.get("body", {}).get("content", "") if isinstance(data, dict) else ""
# Strip HTML tags for readability if HTML
if isinstance(data, dict) and data.get("body", {}).get("contentType") == "html":
import re
body_content = re.sub(r"<[^>]+>", " ", body_content)
body_content = re.sub(r"\s+", " ", body_content).strip()
return json.dumps({
"id": data.get("id") if isinstance(data, dict) else None,
"subject": data.get("subject") if isinstance(data, dict) else None,
"from": data.get("from", {}).get("emailAddress", {}).get("address") if isinstance(data, dict) else None,
"to": [r.get("emailAddress", {}).get("address") for r in data.get("toRecipients", [])] if isinstance(data, dict) else [],
"receivedDateTime": data.get("receivedDateTime") if isinstance(data, dict) else None,
"body": body_content[:16000],
"isRead": data.get("isRead") if isinstance(data, dict) else None,
})
[docs]
async def microsoft_outlook_send(
to: str | list[str],
subject: str,
body: str,
content_type: str = "Text",
ctx: ToolContext | None = None,
) -> str:
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
recipients = to if isinstance(to, list) else [to]
to_recipients = [{"emailAddress": {"address": addr}} for addr in recipients]
payload = {
"message": {
"subject": subject,
"body": {"contentType": content_type, "content": body},
"toRecipients": to_recipients,
},
}
data = await _ms_request("POST", "/me/sendMail", token, json_body=payload)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
return json.dumps({"status": "sent", "to": recipients, "subject": subject})
# ---------------------------------------------------------------------------
# Calendar handlers
# ---------------------------------------------------------------------------
[docs]
async def microsoft_calendar_list(
start_time: str = "",
end_time: str = "",
limit: int = 20,
ctx: ToolContext | None = None,
) -> str:
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
if start_time and end_time:
endpoint = f"/me/calendarView"
params: dict[str, Any] = {
"startDateTime": start_time,
"endDateTime": end_time,
"$top": min(limit, 100),
"$orderby": "start/dateTime",
}
else:
endpoint = "/me/events"
params = {
"$top": min(limit, 100),
"$orderby": "start/dateTime",
}
data = await _ms_request("GET", endpoint, token, params=params)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
events = []
for e in data.get("value", []) if isinstance(data, dict) else []:
events.append({
"id": e.get("id"),
"subject": e.get("subject"),
"bodyPreview": (e.get("bodyPreview") or "")[:300],
"start": e.get("start", {}).get("dateTime"),
"startTimeZone": e.get("start", {}).get("timeZone"),
"end": e.get("end", {}).get("dateTime"),
"endTimeZone": e.get("end", {}).get("timeZone"),
"location": e.get("location", {}).get("displayName"),
"isAllDay": e.get("isAllDay"),
"webLink": e.get("webLink"),
"attendees": [
a.get("emailAddress", {}).get("address")
for a in e.get("attendees", [])[:10]
],
})
return json.dumps({"count": len(events), "events": events})
[docs]
async def microsoft_calendar_create(
subject: str,
start_time: str,
start_timezone: str = "UTC",
end_time: str = "",
end_timezone: str = "UTC",
body: str = "",
location: str = "",
attendees: list[str] | None = None,
is_all_day: bool = False,
ctx: ToolContext | None = None,
) -> str:
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
event_body: dict[str, Any] = {
"subject": subject,
"start": {"dateTime": start_time, "timeZone": start_timezone},
"end": {"dateTime": end_time or start_time, "timeZone": end_timezone},
"isAllDay": is_all_day,
}
if body:
event_body["body"] = {"contentType": "Text", "content": body}
if location:
event_body["location"] = {"displayName": location}
if attendees:
event_body["attendees"] = [
{"emailAddress": {"address": addr}, "type": "required"}
for addr in attendees
]
data = await _ms_request("POST", "/me/events", token, json_body=event_body)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
return json.dumps({
"status": "created",
"id": data.get("id") if isinstance(data, dict) else None,
"subject": data.get("subject") if isinstance(data, dict) else subject,
"webLink": data.get("webLink") if isinstance(data, dict) else None,
"start": data.get("start") if isinstance(data, dict) else start_time,
"end": data.get("end") if isinstance(data, dict) else end_time,
})
# ---------------------------------------------------------------------------
# TOOLS registration
# ---------------------------------------------------------------------------
TOOLS = [
{
"name": "microsoft_onedrive_list",
"description": "List files and folders in the user's OneDrive. Supports path browsing and search.",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Folder path to list (e.g. 'Documents/Reports'). Empty for root."},
"search": {"type": "string", "description": "Search query to find files by name or content"},
"limit": {"type": "integer", "description": "Max results (default 20)"},
},
},
"handler": microsoft_onedrive_list,
},
{
"name": "microsoft_onedrive_read",
"description": "Read the contents of a file from OneDrive by item ID or path.",
"parameters": {
"type": "object",
"properties": {
"item_id": {"type": "string", "description": "OneDrive item ID"},
"path": {"type": "string", "description": "File path (e.g. 'Documents/notes.txt')"},
},
},
"handler": microsoft_onedrive_read,
},
{
"name": "microsoft_onedrive_upload",
"description": "Upload a file to OneDrive.",
"parameters": {
"type": "object",
"properties": {
"name": {"type": "string", "description": "File name"},
"content": {"type": "string", "description": "File content (text)"},
"folder_path": {"type": "string", "description": "Destination folder path (empty for root)"},
},
"required": ["name", "content"],
},
"handler": microsoft_onedrive_upload,
},
{
"name": "microsoft_outlook_list",
"description": "List emails from the user's Outlook mailbox with optional search and folder filter.",
"parameters": {
"type": "object",
"properties": {
"folder": {"type": "string", "description": "Mail folder (default: 'inbox'). Options: inbox, drafts, sentitems, deleteditems"},
"search": {"type": "string", "description": "Search query"},
"limit": {"type": "integer", "description": "Max messages (default 15)"},
"skip": {"type": "integer", "description": "Number of messages to skip (for pagination)"},
},
},
"handler": microsoft_outlook_list,
},
{
"name": "microsoft_outlook_read",
"description": "Read a specific Outlook email by message ID.",
"parameters": {
"type": "object",
"properties": {
"message_id": {"type": "string", "description": "Outlook message ID"},
},
"required": ["message_id"],
},
"handler": microsoft_outlook_read,
},
{
"name": "microsoft_outlook_send",
"description": "Send an email from the user's Outlook account.",
"parameters": {
"type": "object",
"properties": {
"to": {
"description": "Recipient email address(es). String or array of strings.",
"oneOf": [
{"type": "string"},
{"type": "array", "items": {"type": "string"}},
],
},
"subject": {"type": "string", "description": "Email subject"},
"body": {"type": "string", "description": "Email body"},
"content_type": {"type": "string", "enum": ["Text", "HTML"], "description": "Body format (default: Text)"},
},
"required": ["to", "subject", "body"],
},
"handler": microsoft_outlook_send,
},
{
"name": "microsoft_calendar_list",
"description": "List events from the user's Outlook/Microsoft calendar. Optionally filter by time range.",
"parameters": {
"type": "object",
"properties": {
"start_time": {"type": "string", "description": "Start of time range (ISO 8601)"},
"end_time": {"type": "string", "description": "End of time range (ISO 8601)"},
"limit": {"type": "integer", "description": "Max events (default 20)"},
},
},
"handler": microsoft_calendar_list,
},
{
"name": "microsoft_calendar_create",
"description": "Create a new event on the user's Outlook/Microsoft calendar.",
"parameters": {
"type": "object",
"properties": {
"subject": {"type": "string", "description": "Event subject/title"},
"start_time": {"type": "string", "description": "Start time (ISO 8601, e.g. '2025-03-15T10:00:00')"},
"start_timezone": {"type": "string", "description": "Start timezone (default: UTC)"},
"end_time": {"type": "string", "description": "End time (ISO 8601)"},
"end_timezone": {"type": "string", "description": "End timezone (default: UTC)"},
"body": {"type": "string", "description": "Event description"},
"location": {"type": "string", "description": "Event location"},
"attendees": {"type": "array", "items": {"type": "string"}, "description": "Attendee email addresses"},
"is_all_day": {"type": "boolean", "description": "Whether this is an all-day event"},
},
"required": ["subject", "start_time"],
},
"handler": microsoft_calendar_create,
},
]