"""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 jsonutil as json
import logging
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:
"""Perform an authenticated HTTP request against the Microsoft Graph API.
Central low-level helper that every Graph tool in this module routes
through. It attaches the OAuth bearer token, builds the request URL (either
relative to the v1.0 Graph base or a fully-qualified URL such as a Graph
``@odata.nextLink``/download URL), enforces that absolute URLs target only
``graph.microsoft.com``, and normalizes the response into a parsed JSON
object, raw text, or an ``{"error": ...}`` mapping.
It opens a transient :class:`aiohttp.ClientSession`, issues the request,
treats HTTP 204 as a success sentinel, surfaces any HTTP >= 400 status as a
truncated error dict, and otherwise parses the body with the project
``jsonutil`` (aliased ``json``), falling back to truncated raw text on a
decode failure. No Redis, KG, LLM, or event-bus interaction occurs here.
Called by every OneDrive/Outlook/Calendar handler in this module
(``microsoft_onedrive_list``, ``microsoft_onedrive_read``,
``microsoft_onedrive_upload``, ``microsoft_outlook_list``,
``microsoft_outlook_read``, ``microsoft_outlook_send``,
``microsoft_calendar_list``, ``microsoft_calendar_create``) and exercised
directly by ``tests/test_security_remediation.py`` to verify host pinning.
Args:
method: HTTP verb (e.g. ``"GET"``, ``"POST"``, ``"PUT"``).
path: Graph path beginning with ``/`` (appended to the v1.0 base) or a
full ``http(s)://`` URL, which must be on ``graph.microsoft.com``.
token: OAuth bearer access token for the calling user's Microsoft
account.
params: Optional query-string parameters (e.g. Graph ``$top``/``$select``).
json_body: Optional JSON-serializable request body.
data: Optional raw byte body (used for OneDrive content uploads); takes
its content type from ``extra_headers``.
extra_headers: Optional headers merged over the ``Authorization`` header
(e.g. ``Content-Type``).
Returns:
dict[str, Any] | list[Any] | str: The parsed JSON response, a status
dict for 204, an ``{"error": ..., "detail": ...}`` dict for a blocked
host or HTTP error, or truncated raw text when the body is not JSON.
"""
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:
"""Resolve a valid Microsoft OAuth access token for the current user.
Thin wrapper that validates the tool context has Redis available, then
delegates to ``oauth_manager.require_oauth_token`` to fetch (and refresh as
needed) the per-user ``"microsoft"`` access token. Used at the top of every
handler in this module so that a missing connection can be surfaced to the
user as a connect link rather than a hard failure.
Through ``require_oauth_token`` this drives the ``OAuthManager``, which reads
the user's stored token from Redis (keyed by ``ctx.user_id``) and, when no
token exists, generates an OAuth connect URL and raises ``OAuthNotConnected``.
It is called by ``microsoft_onedrive_list``, ``microsoft_onedrive_read``,
``microsoft_onedrive_upload``, ``microsoft_outlook_list``,
``microsoft_outlook_read``, ``microsoft_outlook_send``,
``microsoft_calendar_list``, and ``microsoft_calendar_create`` (each catching
``OAuthNotConnected`` to return the connect link to the model).
Args:
ctx: The active :class:`ToolContext`, carrying ``user_id`` and the Redis
client. May be ``None`` when invoked outside a normal tool run.
Returns:
str: A valid Microsoft Graph bearer access token.
Raises:
RuntimeError: If ``ctx`` is ``None`` or its Redis client is unavailable.
oauth_manager.OAuthNotConnected: If the user has not connected a
Microsoft account (carries a connect URL).
"""
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:
"""List files and folders in the calling user's OneDrive.
Tool handler for ``microsoft_onedrive_list``. Browses a folder, searches by
name/content, or lists the drive root, returning a compact JSON summary of
each child item (id, name, size, modified time, web URL, and folder/file
type metadata). Path and search inputs are validated to reject newlines and
``..`` traversal before building the Graph endpoint.
Resolves the user's token via ``_get_token`` (returning the connect-link
message on ``OAuthNotConnected``) and queries the Graph search,
``/children``, or root-children endpoint through ``_ms_request``. Registered
in this module's ``TOOLS`` list and dispatched by ``tool_loader.py`` (via
``getattr(module, "TOOLS")`` → ``ToolDefinition.handler``); no direct
in-repo Python callers were found.
Args:
path: OneDrive folder path to list (e.g. ``"Documents/Reports"``); empty
lists the drive root. Ignored when ``search`` is provided.
search: Query to find files by name or content. Takes precedence over
``path`` when non-empty.
limit: Maximum number of items to return (capped at 200 by the API call).
ctx: The active :class:`ToolContext` providing user identity and Redis.
Returns:
str: A JSON string of the form ``{"count": int, "items": [...]}`` on
success, or a JSON ``{"error": ...}`` / connect-link string on failure.
"""
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:
"""Read a OneDrive file's metadata and text content by item ID or path.
Tool handler for ``microsoft_onedrive_read``. Fetches the item's Graph
metadata, then (if the item exposes a pre-authenticated
``@microsoft.graph.downloadUrl``) downloads the body and decodes it as UTF-8,
truncating to 16 000 characters. Either ``item_id`` or ``path`` must be
supplied, and both are validated against traversal/newline injection.
Resolves the token via ``_get_token`` (returning the connect-link message on
``OAuthNotConnected``), reads metadata through ``_ms_request``, and downloads
the content with a separate transient :class:`aiohttp.ClientSession` against
the Graph-issued download URL. Registered in this module's ``TOOLS`` list and
dispatched by ``tool_loader.py``; no direct in-repo Python callers were found.
Args:
item_id: OneDrive item ID; mutually exclusive with ``path``. Must not
contain ``/``, ``..``, or newlines.
path: File path (e.g. ``"Documents/notes.txt"``); used when ``item_id``
is empty. Must not contain newlines or ``..``.
ctx: The active :class:`ToolContext` providing user identity and Redis.
Returns:
str: A JSON string ``{"metadata": {...}, "content": str}`` on success,
or a JSON ``{"error": ...}`` / connect-link string on failure.
"""
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:
"""Upload a text file to the user's OneDrive.
Tool handler for ``microsoft_onedrive_upload``. Writes ``content`` to a new
or overwritten file at ``name`` (optionally within ``folder_path``) using a
simple Graph content ``PUT``. File name and folder are validated against
newline/``..`` injection before the endpoint is built.
Resolves the token via ``_get_token`` (returning the connect-link message on
``OAuthNotConnected``) and performs the upload through ``_ms_request`` with
the UTF-8-encoded body and an ``application/octet-stream`` content type.
Registered in this module's ``TOOLS`` list and dispatched by
``tool_loader.py``; no direct in-repo Python callers were found.
Args:
name: Destination file name. Must not contain newlines or ``..``.
content: Text content to upload; encoded as UTF-8 bytes.
folder_path: Optional destination folder path; empty uploads to the
drive root. Must not contain newlines or ``..``.
ctx: The active :class:`ToolContext` providing user identity and Redis.
Returns:
str: A JSON string ``{"status": "uploaded", "id": ..., "name": ...,
"webUrl": ..., "size": ...}`` on success, or a JSON ``{"error": ...}`` /
connect-link string on failure.
"""
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:
"""List messages from one of the user's Outlook mail folders.
Tool handler for ``microsoft_outlook_list``. Returns a compact JSON summary
of messages (id, subject, from, to, received time, body preview, read flag,
attachment flag) ordered newest-first, with optional Graph ``$search`` and
pagination via ``skip``.
Resolves the token via ``_get_token`` (returning the connect-link message on
``OAuthNotConnected``) and queries ``/me/mailFolders/{folder}/messages``
through ``_ms_request`` using ``$top``/``$skip``/``$select``/``$orderby``
parameters. Registered in this module's ``TOOLS`` list and dispatched by
``tool_loader.py``; no direct in-repo Python callers were found.
Args:
folder: Mail folder to read (``"inbox"``, ``"drafts"``, ``"sentitems"``,
``"deleteditems"``, etc.).
search: Optional free-text Graph search query.
limit: Maximum messages to return (capped at 50 by the API call).
skip: Number of messages to skip for pagination.
ctx: The active :class:`ToolContext` providing user identity and Redis.
Returns:
str: A JSON string ``{"count": int, "messages": [...]}`` on success, or
a JSON ``{"error": ...}`` / connect-link string on failure.
"""
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:
"""Read a single Outlook email by message ID, returning its full body.
Tool handler for ``microsoft_outlook_read``. Fetches the message via Graph
and returns headers plus the body; HTML bodies are stripped of tags and
whitespace-collapsed for readability, and the body is truncated to 16 000
characters.
Resolves the token via ``_get_token`` (returning the connect-link message on
``OAuthNotConnected``) and fetches ``/me/messages/{message_id}`` through
``_ms_request``. Registered in this module's ``TOOLS`` list and dispatched by
``tool_loader.py``; no direct in-repo Python callers were found.
Args:
message_id: The Outlook message ID to read.
ctx: The active :class:`ToolContext` providing user identity and Redis.
Returns:
str: A JSON string with ``id``, ``subject``, ``from``, ``to``,
``receivedDateTime``, ``body``, and ``isRead`` on success, or a JSON
``{"error": ...}`` / connect-link string on failure.
"""
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:
"""Send an email from the user's Outlook account.
Tool handler for ``microsoft_outlook_send``. Normalizes one or more
recipients into Graph ``toRecipients`` entries and submits the message via
the Graph ``sendMail`` action, which sends immediately (no draft step).
Resolves the token via ``_get_token`` (returning the connect-link message on
``OAuthNotConnected``) and POSTs the message payload to ``/me/sendMail``
through ``_ms_request``. Registered in this module's ``TOOLS`` list and
dispatched by ``tool_loader.py``; no direct in-repo Python callers were found.
Args:
to: A single recipient address or a list of addresses.
subject: Email subject line.
body: Email body content.
content_type: Body format, ``"Text"`` (default) or ``"HTML"``.
ctx: The active :class:`ToolContext` providing user identity and Redis.
Returns:
str: A JSON string ``{"status": "sent", "to": [...], "subject": ...}``
on success, or a JSON ``{"error": ...}`` / connect-link string on
failure.
"""
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:
"""List events from the user's Microsoft/Outlook calendar.
Tool handler for ``microsoft_calendar_list``. When both ``start_time`` and
``end_time`` are supplied it uses the Graph ``calendarView`` (which expands
recurring events within the window); otherwise it lists raw ``events``.
Returns a compact JSON summary per event (subject, body preview, start/end
with time zones, location, all-day flag, web link, and up to ten attendee
addresses).
Resolves the token via ``_get_token`` (returning the connect-link message on
``OAuthNotConnected``) and queries the chosen endpoint through
``_ms_request`` ordered by start time. Registered in this module's ``TOOLS``
list and dispatched by ``tool_loader.py``; no direct in-repo Python callers
were found.
Args:
start_time: ISO 8601 start of the time range; with ``end_time`` switches
to ``calendarView``.
end_time: ISO 8601 end of the time range.
limit: Maximum events to return (capped at 100 by the API call).
ctx: The active :class:`ToolContext` providing user identity and Redis.
Returns:
str: A JSON string ``{"count": int, "events": [...]}`` on success, or a
JSON ``{"error": ...}`` / connect-link string on failure.
"""
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 = "/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:
"""Create a new event on the user's Microsoft/Outlook calendar.
Tool handler for ``microsoft_calendar_create``. Builds a Graph event payload
from the supplied subject, start/end times and time zones, and optional body,
location, attendees, and all-day flag (each attendee added as a required
invitee). When ``end_time`` is empty it defaults to ``start_time``.
Resolves the token via ``_get_token`` (returning the connect-link message on
``OAuthNotConnected``) and POSTs the event to ``/me/events`` through
``_ms_request``. Registered in this module's ``TOOLS`` list and dispatched by
``tool_loader.py``; no direct in-repo Python callers were found.
Args:
subject: Event title.
start_time: ISO 8601 start time (e.g. ``"2025-03-15T10:00:00"``).
start_timezone: Start time zone (default ``"UTC"``).
end_time: ISO 8601 end time; defaults to ``start_time`` when empty.
end_timezone: End time zone (default ``"UTC"``).
body: Optional plain-text event description.
location: Optional location display name.
attendees: Optional list of attendee email addresses (added as required).
is_all_day: Whether the event spans the whole day.
ctx: The active :class:`ToolContext` providing user identity and Redis.
Returns:
str: A JSON string ``{"status": "created", "id": ..., "subject": ...,
"webLink": ..., "start": ..., "end": ...}`` on success, or a JSON
``{"error": ...}`` / connect-link string on failure.
"""
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,
},
]