ef886bb503
| Name | Description | |------------------------------|--------------------------------------------------------------------------------| | Microsoft.CreateDraftEmail | Compose a new draft email in Outlook | | Microsoft.UpdateDraftEmail | Update an existing draft email in Outlook | | Microsoft.CreateAndSendEmail | Create and immediately send a new email in Outlook to the specified recipients | | Microsoft.SendDraftEmail | Send an existing draft email in Outlook | | Microsoft.ReplyToEmail | Reply only to the sender of an existing email in Outlook | | Microsoft.ReplyAllToEmail | Reply to all recipients of an existing email in Outlook | | Microsoft.ListEmails | List emails in the user's mailbox across all folders | | Microsoft.ListEmailsInFolder | List the user's emails in the specified folder |
89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
from typing import Annotated
|
|
|
|
from arcade.sdk import ToolContext, tool
|
|
from arcade.sdk.auth import Microsoft
|
|
from arcade.sdk.errors import ToolExecutionError
|
|
|
|
from arcade_microsoft.client import get_client
|
|
from arcade_microsoft.outlook_mail._utils import (
|
|
fetch_emails,
|
|
prepare_list_emails_request_config,
|
|
remove_none_values,
|
|
)
|
|
from arcade_microsoft.outlook_mail.enums import WellKnownFolderNames
|
|
from arcade_microsoft.outlook_mail.message import Message
|
|
|
|
|
|
@tool(requires_auth=Microsoft(scopes=["Mail.Read"]))
|
|
async def list_emails(
|
|
context: ToolContext,
|
|
limit: Annotated[int, "The number of messages to return. Max is 100. Defaults to 5."] = 5,
|
|
pagination_token: Annotated[
|
|
str | None, "The pagination token to continue a previous request"
|
|
] = None,
|
|
) -> Annotated[dict, "A dictionary containing a list of emails"]:
|
|
"""List emails in the user's mailbox across all folders.
|
|
|
|
Since this tool lists email across all folders, it may return sent items, drafts,
|
|
and other items that are not in the inbox.
|
|
"""
|
|
client = get_client(context.get_auth_token_or_empty())
|
|
request_config = prepare_list_emails_request_config(limit)
|
|
message_builder = client.me.messages
|
|
|
|
response = await fetch_emails(message_builder, pagination_token, request_config)
|
|
messages = [Message.from_sdk(msg).to_dict() for msg in response.value or []]
|
|
pagination_token = response.odata_next_link
|
|
|
|
result = {
|
|
"messages": messages,
|
|
"num_messages": len(messages),
|
|
"pagination_token": pagination_token,
|
|
}
|
|
result = remove_none_values(result)
|
|
return result
|
|
|
|
|
|
@tool(requires_auth=Microsoft(scopes=["Mail.Read"]))
|
|
async def list_emails_in_folder(
|
|
context: ToolContext,
|
|
well_known_folder_name: Annotated[
|
|
WellKnownFolderNames | None,
|
|
"The name of the folder to list emails from. Defaults to None.",
|
|
] = None,
|
|
folder_id: Annotated[
|
|
str | None,
|
|
"The ID of the folder to list emails from if the folder is not a well-known folder. "
|
|
"Defaults to None.",
|
|
] = None,
|
|
limit: Annotated[int, "The number of messages to return. Max is 100. Defaults to 5."] = 5,
|
|
pagination_token: Annotated[
|
|
str | None, "The pagination token to continue a previous request"
|
|
] = None,
|
|
) -> Annotated[
|
|
dict, "A dictionary containing a list of emails and a pagination token, if applicable"
|
|
]:
|
|
"""List the user's emails in the specified folder.
|
|
|
|
Exactly one of `well_known_folder_name` or `folder_id` MUST be provided.
|
|
"""
|
|
if not (bool(well_known_folder_name) ^ bool(folder_id)):
|
|
raise ToolExecutionError(
|
|
message="Exactly one of `well_known_folder_name` or `folder_id` must be provided."
|
|
)
|
|
folder_name = well_known_folder_name.value if well_known_folder_name else folder_id
|
|
client = get_client(context.get_auth_token_or_empty())
|
|
request_config = prepare_list_emails_request_config(limit)
|
|
message_builder = client.me.mail_folders.by_mail_folder_id(folder_name).messages # type: ignore[arg-type]
|
|
|
|
response = await fetch_emails(message_builder, pagination_token, request_config)
|
|
messages = [Message.from_sdk(msg).to_dict() for msg in response.value or []]
|
|
pagination_token = response.odata_next_link
|
|
|
|
result = {
|
|
"messages": messages,
|
|
"num_messages": len(messages),
|
|
"pagination_token": pagination_token,
|
|
}
|
|
result = remove_none_values(result)
|
|
return result
|