app.py•4.66 kB
from datetime import datetime
from typing import Annotated, Literal
from mcp.server.fastmcp import FastMCP
from pydantic import Field
from mcp_email_server.config import (
AccountAttributes,
EmailSettings,
ProviderSettings,
get_settings,
)
from mcp_email_server.emails.dispatcher import dispatch_handler
from mcp_email_server.emails.models import EmailContentBatchResponse, EmailMetadataPageResponse
mcp = FastMCP("email")
@mcp.resource("email://{account_name}")
async def get_account(account_name: str) -> EmailSettings | ProviderSettings | None:
settings = get_settings()
return settings.get_account(account_name, masked=True)
@mcp.tool(description="List all configured email accounts with masked credentials.")
async def list_available_accounts() -> list[AccountAttributes]:
settings = get_settings()
return [account.masked() for account in settings.get_accounts()]
@mcp.tool(description="Add a new email account configuration to the settings.")
async def add_email_account(email: EmailSettings) -> str:
settings = get_settings()
settings.add_email(email)
settings.store()
return f"Successfully added email account '{email.account_name}'"
@mcp.tool(
description="List email metadata (email_id, subject, sender, recipients, date) without body content. Returns email_id for use with get_emails_content."
)
async def list_emails_metadata(
account_name: Annotated[str, Field(description="The name of the email account.")],
page: Annotated[
int,
Field(default=1, description="The page number to retrieve (starting from 1)."),
] = 1,
page_size: Annotated[int, Field(default=10, description="The number of emails to retrieve per page.")] = 10,
before: Annotated[
datetime | None,
Field(default=None, description="Retrieve emails before this datetime (UTC)."),
] = None,
since: Annotated[
datetime | None,
Field(default=None, description="Retrieve emails since this datetime (UTC)."),
] = None,
subject: Annotated[str | None, Field(default=None, description="Filter emails by subject.")] = None,
from_address: Annotated[str | None, Field(default=None, description="Filter emails by sender address.")] = None,
to_address: Annotated[
str | None,
Field(default=None, description="Filter emails by recipient address."),
] = None,
order: Annotated[
Literal["asc", "desc"],
Field(default=None, description="Order emails by field. `asc` or `desc`."),
] = "desc",
) -> EmailMetadataPageResponse:
handler = dispatch_handler(account_name)
return await handler.get_emails_metadata(
page=page,
page_size=page_size,
before=before,
since=since,
subject=subject,
from_address=from_address,
to_address=to_address,
order=order,
)
@mcp.tool(
description="Get the full content (including body) of one or more emails by their email_id. Use list_emails_metadata first to get the email_id."
)
async def get_emails_content(
account_name: Annotated[str, Field(description="The name of the email account.")],
email_ids: Annotated[
list[str],
Field(
description="List of email_id to retrieve (obtained from list_emails_metadata). Can be a single email_id or multiple email_ids."
),
],
) -> EmailContentBatchResponse:
handler = dispatch_handler(account_name)
return await handler.get_emails_content(email_ids)
@mcp.tool(
description="Send an email using the specified account. Recipient should be a list of email addresses.",
)
async def send_email(
account_name: Annotated[str, Field(description="The name of the email account to send from.")],
recipients: Annotated[list[str], Field(description="A list of recipient email addresses.")],
subject: Annotated[str, Field(description="The subject of the email.")],
body: Annotated[str, Field(description="The body of the email.")],
cc: Annotated[
list[str] | None,
Field(default=None, description="A list of CC email addresses."),
] = None,
bcc: Annotated[
list[str] | None,
Field(default=None, description="A list of BCC email addresses."),
] = None,
html: Annotated[
bool,
Field(default=False, description="Whether to send the email as HTML (True) or plain text (False)."),
] = False,
) -> str:
handler = dispatch_handler(account_name)
await handler.send_email(recipients, subject, body, cc, bcc, html)
recipient_str = ", ".join(recipients)
return f"Email sent successfully to {recipient_str}"