yigit/dav-imap-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add tool call logging and reply fields
All checks were successful
Build And Test / publish (push) Successful in 49s
Details

Browse Source
This commit is contained in:
Yigit Colakoglu
2026-01-01 15:24:06 -08:00
parent 7966a4302d
commit 767f076048
7 changed files with 335 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
src/tools/email_tools.py
View File

@@ -2,18 +2,21 @@ from typing import Optional
from fastmcp import FastMCP
from services.email_service import EmailService
from tools.logging_utils import log_tool_call
def register_email_tools(mcp: FastMCP, service: EmailService):
"""Register all email-related MCP tools."""
@mcp.tool(description="List all mailboxes/folders in the email account. Returns name, path, message count, and unread count for each mailbox.")
@log_tool_call
def list_mailboxes() -> list[dict]:
"""List all IMAP mailboxes/folders."""
mailboxes = service.list_mailboxes()
return [m.model_dump() for m in mailboxes]
@mcp.tool(description="List emails in a mailbox with pagination. Returns email summaries including subject, from, date, and read status.")
@log_tool_call
def list_emails(
mailbox: str = "INBOX",
limit: int = 50,
@@ -33,6 +36,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
return result.model_dump()
@mcp.tool(description="List draft emails in the Drafts mailbox with pagination.")
@log_tool_call
def list_drafts(
mailbox: Optional[str] = None,
limit: int = 50,
@@ -52,6 +56,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
return result.model_dump()
@mcp.tool(description="Read a specific email by ID with full body content and attachment information.")
@log_tool_call
def read_email(
mailbox: str,
email_id: str,
@@ -69,6 +74,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
return result.model_dump() if result else None
@mcp.tool(description="Search emails in a mailbox using various criteria like subject, sender, or body content.")
@log_tool_call
def search_emails(
query: str,
mailbox: str = "INBOX",
@@ -94,6 +100,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
return result.model_dump()
@mcp.tool(description="Move an email from one mailbox/folder to another.")
@log_tool_call
def move_email(
email_id: str,
source_mailbox: str,
@@ -111,6 +118,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
return result.model_dump()
@mcp.tool(description="Delete an email, either moving it to trash or permanently deleting it.")
@log_tool_call
def delete_email(
email_id: str,
mailbox: str,
@@ -128,6 +136,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
return result.model_dump()
@mcp.tool(description="Delete a drafted email by ID, optionally permanently.")
@log_tool_call
def delete_draft(
email_id: str,
mailbox: Optional[str] = None,
@@ -145,33 +154,53 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
return result.model_dump()
@mcp.tool(description="Save a new draft email to the Drafts mailbox.")
@log_tool_call
def save_draft(
to: list[str],
subject: str,
body: str,
to: Optional[list[str]] = None,
subject: Optional[str] = None,
body: Optional[str] = None,
cc: Optional[list[str]] = None,
bcc: Optional[list[str]] = None,
reply_to: Optional[str] = None,
html_body: Optional[str] = None,
mailbox: Optional[str] = None,
reply_to_email_id: Optional[str] = None,
reply_mailbox: Optional[str] = None,
reply_all: bool = False,
) -> dict:
"""
Save a new email draft.
Args:
to: List of recipient email addresses
subject: Email subject line
body: Plain text email body
to: List of recipient email addresses (required unless reply_to_email_id is set)
subject: Email subject line (required unless reply_to_email_id is set)
body: Plain text email body (required unless reply_to_email_id is set)
cc: List of CC recipients (optional)
bcc: List of BCC recipients (optional)
reply_to: Reply-to address (optional)
html_body: HTML version of the email body (optional)
mailbox: Drafts mailbox/folder override (default: auto-detect)
reply_to_email_id: Email ID to reply to (optional)
reply_mailbox: Mailbox containing the reply_to_email_id (default: INBOX)
reply_all: Whether to include original recipients when replying (default: False)
"""
result = service.save_draft(to, subject, body, cc, bcc, reply_to, html_body, mailbox)
result = service.save_draft(
to,
subject,
body,
cc,
bcc,
reply_to,
html_body,
mailbox,
reply_to_email_id,
reply_mailbox,
reply_all,
)
return result.model_dump()
@mcp.tool(description="Edit an existing draft email. Only provided fields will be modified.")
@log_tool_call
def edit_draft(
email_id: str,
mailbox: Optional[str] = None,
@@ -182,6 +211,9 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
bcc: Optional[list[str]] = None,
reply_to: Optional[str] = None,
html_body: Optional[str] = None,
reply_to_email_id: Optional[str] = None,
reply_mailbox: Optional[str] = None,
reply_all: bool = False,
) -> dict:
"""
Update an existing draft email.
@@ -196,6 +228,9 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
bcc: List of BCC recipients (optional)
reply_to: Reply-to address (optional)
html_body: HTML version of the email body (optional)
reply_to_email_id: Email ID to reply to (optional)
reply_mailbox: Mailbox containing the reply_to_email_id (default: INBOX)
reply_all: Whether to include original recipients when replying (default: False)
"""
result = service.update_draft(
email_id,
@@ -207,34 +242,44 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
bcc,
reply_to,
html_body,
reply_to_email_id,
reply_mailbox,
reply_all,
)
return result.model_dump()
@mcp.tool(description="Send a new email via SMTP. Supports plain text and HTML content, CC, BCC, reply-to, and custom sender.")
@log_tool_call
async def send_email(
to: list[str],
subject: str,
body: str,
to: Optional[list[str]] = None,
subject: Optional[str] = None,
body: Optional[str] = None,
cc: Optional[list[str]] = None,
bcc: Optional[list[str]] = None,
reply_to: Optional[str] = None,
html_body: Optional[str] = None,
sender_email: Optional[str] = None,
sender_name: Optional[str] = None,
reply_to_email_id: Optional[str] = None,
reply_mailbox: Optional[str] = None,
reply_all: bool = False,
) -> dict:
"""
Send a new email.
Args:
to: List of recipient email addresses
subject: Email subject line
body: Plain text email body
to: List of recipient email addresses (required unless reply_to_email_id is set)
subject: Email subject line (required unless reply_to_email_id is set)
body: Plain text email body (required unless reply_to_email_id is set)
cc: List of CC recipients (optional)
bcc: List of BCC recipients (optional)
reply_to: Reply-to address (optional)
html_body: HTML version of the email body (optional)
sender_email: Sender email address (optional, defaults to SMTP_FROM_EMAIL)
sender_name: Sender display name (optional, defaults to SMTP_FROM_NAME)
reply_to_email_id: Email ID to reply to (optional)
reply_mailbox: Mailbox containing the reply_to_email_id (default: INBOX)
reply_all: Whether to include original recipients when replying (default: False)
"""
result = await service.send_email(
to,
@@ -246,52 +291,16 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
html_body,
sender_email,
sender_name,
)
return result.model_dump()
@mcp.tool(description="Reply to an existing email by ID, with optional reply-all behavior.")
async def reply_email(
email_id: str,
body: str,
mailbox: str = "INBOX",
reply_all: bool = False,
cc: Optional[list[str]] = None,
bcc: Optional[list[str]] = None,
reply_to: Optional[str] = None,
html_body: Optional[str] = None,
sender_email: Optional[str] = None,
sender_name: Optional[str] = None,
) -> dict:
"""
Reply to an existing email.
Args:
email_id: The unique ID of the email to reply to
body: Plain text email body
mailbox: The mailbox containing the email (default: INBOX)
reply_all: Whether to include original recipients (default: False)
cc: List of CC recipients (optional)
bcc: List of BCC recipients (optional)
reply_to: Reply-to address for the reply (optional)
html_body: HTML version of the email body (optional)
sender_email: Sender email address (optional, defaults to SMTP_FROM_EMAIL)
sender_name: Sender display name (optional, defaults to SMTP_FROM_NAME)
"""
result = await service.reply_email(
mailbox,
email_id,
body,
None,
None,
reply_to_email_id,
reply_mailbox,
reply_all,
cc,
bcc,
reply_to,
html_body,
sender_email,
sender_name,
)
return result.model_dump()
@mcp.tool(description="Set or remove IMAP flags on an email. Standard flags: \\Seen, \\Answered, \\Flagged, \\Deleted, \\Draft. Custom keywords are also supported.")
@log_tool_call
def set_email_flags(
email_id: str,
mailbox: str,
@@ -311,6 +320,7 @@ def register_email_tools(mcp: FastMCP, service: EmailService):
return result.model_dump()
@mcp.tool(description="Unsubscribe from a mailing list. Parses List-Unsubscribe headers and attempts automatic unsubscribe via HTTP or provides mailto instructions.")
@log_tool_call
async def unsubscribe_email(
email_id: str,
mailbox: str = "INBOX",