jason/google-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ruff

Browse Source
This commit is contained in:
Taylor Wilsdon
2026-01-30 10:20:05 -05:00
parent ead1229874 e12599773c
commit 054780664d
45 changed files with 7847 additions and 1293 deletions
Download Patch File Download Diff File Expand all files Collapse all files

244
gmail/gmail_tools.py
View File

@@ -8,10 +8,15 @@ import logging
import asyncio
import base64
import ssl
import mimetypes
from pathlib import Path
from html.parser import HTMLParser
from typing import Optional, List, Dict, Literal, Any
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
from fastapi import Body
from pydantic import Field
@@ -31,6 +36,7 @@ logger = logging.getLogger(__name__)
GMAIL_BATCH_SIZE = 25
GMAIL_REQUEST_DELAY = 0.1
HTML_BODY_TRUNCATE_LIMIT = 20000
GMAIL_METADATA_HEADERS = ["Subject", "From", "To", "Cc", "Message-ID", "Date"]
class _HTMLTextExtractor(HTMLParser):
@@ -214,9 +220,12 @@ def _extract_headers(payload: dict, header_names: List[str]) -> Dict[str, str]:
Dict mapping header names to their values
"""
headers = {}
target_headers = {name.lower(): name for name in header_names}
for header in payload.get("headers", []):
if header["name"] in header_names:
headers[header["name"]] = header["value"]
header_name_lower = header["name"].lower()
if header_name_lower in target_headers:
# Store using the original requested casing
headers[target_headers[header_name_lower]] = header["value"]
return headers
@@ -232,9 +241,10 @@ def _prepare_gmail_message(
body_format: Literal["plain", "html"] = "plain",
from_email: Optional[str] = None,
from_name: Optional[str] = None,
attachments: Optional[List[Dict[str, str]]] = None,
) -> tuple[str, Optional[str]]:
"""
Prepare a Gmail message with threading support.
Prepare a Gmail message with threading and attachment support.
Args:
subject: Email subject
@@ -248,6 +258,7 @@ def _prepare_gmail_message(
body_format: Content type for the email body ('plain' or 'html')
from_email: Optional sender email address
from_name: Optional sender display name (e.g., "Peter Hartree")
attachments: Optional list of attachments. Each can have 'path' (file path) OR 'content' (base64) + 'filename'
Returns:
Tuple of (raw_message, thread_id) where raw_message is base64 encoded
@@ -262,7 +273,82 @@ def _prepare_gmail_message(
if normalized_format not in {"plain", "html"}:
raise ValueError("body_format must be either 'plain' or 'html'.")
message = MIMEText(body, normalized_format)
# Use multipart if attachments are provided
if attachments:
message = MIMEMultipart()
message.attach(MIMEText(body, normalized_format))
# Process attachments
for attachment in attachments:
file_path = attachment.get("path")
filename = attachment.get("filename")
content_base64 = attachment.get("content")
mime_type = attachment.get("mime_type")
try:
# If path is provided, read and encode the file
if file_path:
path_obj = Path(file_path)
if not path_obj.exists():
logger.error(f"File not found: {file_path}")
continue
# Read file content
with open(path_obj, "rb") as f:
file_data = f.read()
# Use provided filename or extract from path
if not filename:
filename = path_obj.name
# Auto-detect MIME type if not provided
if not mime_type:
mime_type, _ = mimetypes.guess_type(str(path_obj))
if not mime_type:
mime_type = "application/octet-stream"
# If content is provided (base64), decode it
elif content_base64:
if not filename:
logger.warning("Skipping attachment: missing filename")
continue
file_data = base64.b64decode(content_base64)
if not mime_type:
mime_type = "application/octet-stream"
else:
logger.warning("Skipping attachment: missing both path and content")
continue
# Create MIME attachment
main_type, sub_type = mime_type.split("/", 1)
part = MIMEBase(main_type, sub_type)
part.set_payload(file_data)
encoders.encode_base64(part)
# Sanitize filename to prevent header injection and ensure valid quoting
safe_filename = (
(filename or "")
.replace("\r", "")
.replace("\n", "")
.replace("\\", "\\\\")
.replace('"', r"\"")
)
part.add_header(
"Content-Disposition", f'attachment; filename="{safe_filename}"'
)
message.attach(part)
logger.info(f"Attached file: {filename} ({len(file_data)} bytes)")
except Exception as e:
logger.error(f"Failed to attach {filename or file_path}: {e}")
continue
else:
message = MIMEText(body, normalized_format)
message["Subject"] = reply_subject
# Add sender if provided
@@ -462,7 +548,7 @@ async def get_gmail_message_content(
user_google_email (str): The user's Google email address. Required.
Returns:
str: The message details including subject, sender, recipients (To, Cc), and body content.
str: The message details including subject, sender, date, Message-ID, recipients (To, Cc), and body content.
"""
logger.info(
f"[get_gmail_message_content] Invoked. Message ID: '{message_id}', Email: '{user_google_email}'"
@@ -478,19 +564,19 @@ async def get_gmail_message_content(
userId="me",
id=message_id,
format="metadata",
metadataHeaders=["Subject", "From", "To", "Cc"],
metadataHeaders=GMAIL_METADATA_HEADERS,
)
.execute
)
headers = {
h["name"]: h["value"]
for h in message_metadata.get("payload", {}).get("headers", [])
}
headers = _extract_headers(
message_metadata.get("payload", {}), GMAIL_METADATA_HEADERS
)
subject = headers.get("Subject", "(no subject)")
sender = headers.get("From", "(unknown sender)")
to = headers.get("To", "")
cc = headers.get("Cc", "")
rfc822_msg_id = headers.get("Message-ID", "")
# Now fetch the full message to get the body parts
message_full = await asyncio.to_thread(
@@ -519,8 +605,12 @@ async def get_gmail_message_content(
content_lines = [
f"Subject: {subject}",
f"From: {sender}",
f"Date: {headers.get('Date', '(unknown date)')}",
]
if rfc822_msg_id:
content_lines.append(f"Message-ID: {rfc822_msg_id}")
if to:
content_lines.append(f"To: {to}")
if cc:
@@ -563,7 +653,7 @@ async def get_gmail_messages_content_batch(
format (Literal["full", "metadata"]): Message format. "full" includes body, "metadata" only headers.
Returns:
str: A formatted list of message contents including subject, sender, recipients (To, Cc), and body (if full format).
str: A formatted list of message contents including subject, sender, date, Message-ID, recipients (To, Cc), and body (if full format).
"""
logger.info(
f"[get_gmail_messages_content_batch] Invoked. Message count: {len(message_ids)}, Email: '{user_google_email}'"
@@ -596,7 +686,7 @@ async def get_gmail_messages_content_batch(
userId="me",
id=mid,
format="metadata",
metadataHeaders=["Subject", "From", "To", "Cc"],
metadataHeaders=GMAIL_METADATA_HEADERS,
)
)
else:
@@ -628,7 +718,7 @@ async def get_gmail_messages_content_batch(
userId="me",
id=mid,
format="metadata",
metadataHeaders=["Subject", "From", "To", "Cc"],
metadataHeaders=GMAIL_METADATA_HEADERS,
)
.execute
)
@@ -679,15 +769,20 @@ async def get_gmail_messages_content_batch(
payload = message.get("payload", {})
if format == "metadata":
headers = _extract_headers(payload, ["Subject", "From", "To", "Cc"])
headers = _extract_headers(payload, GMAIL_METADATA_HEADERS)
subject = headers.get("Subject", "(no subject)")
sender = headers.get("From", "(unknown sender)")
to = headers.get("To", "")
cc = headers.get("Cc", "")
rfc822_msg_id = headers.get("Message-ID", "")
msg_output = (
f"Message ID: {mid}\nSubject: {subject}\nFrom: {sender}\n"
f"Date: {headers.get('Date', '(unknown date)')}\n"
)
if rfc822_msg_id:
msg_output += f"Message-ID: {rfc822_msg_id}\n"
if to:
msg_output += f"To: {to}\n"
if cc:
@@ -697,11 +792,12 @@ async def get_gmail_messages_content_batch(
output_messages.append(msg_output)
else:
# Full format - extract body too
headers = _extract_headers(payload, ["Subject", "From", "To", "Cc"])
headers = _extract_headers(payload, GMAIL_METADATA_HEADERS)
subject = headers.get("Subject", "(no subject)")
sender = headers.get("From", "(unknown sender)")
to = headers.get("To", "")
cc = headers.get("Cc", "")
rfc822_msg_id = headers.get("Message-ID", "")
# Extract both text and HTML bodies using enhanced helper function
bodies = _extract_message_bodies(payload)
@@ -713,7 +809,11 @@ async def get_gmail_messages_content_batch(
msg_output = (
f"Message ID: {mid}\nSubject: {subject}\nFrom: {sender}\n"
f"Date: {headers.get('Date', '(unknown date)')}\n"
)
if rfc822_msg_id:
msg_output += f"Message-ID: {rfc822_msg_id}\n"
if to:
msg_output += f"To: {to}\n"
if cc:
@@ -894,7 +994,12 @@ async def send_gmail_message(
cc: Optional[str] = Body(None, description="Optional CC email address."),
bcc: Optional[str] = Body(None, description="Optional BCC email address."),
from_name: Optional[str] = Body(
None, description="Optional sender display name (e.g., 'Peter Hartree'). If provided, the From header will be formatted as 'Name <email>'."
None,
description="Optional sender display name (e.g., 'Peter Hartree'). If provided, the From header will be formatted as 'Name <email>'.",
),
from_email: Optional[str] = Body(
None,
description="Optional 'Send As' alias email address. Must be configured in Gmail settings (Settings > Accounts > Send mail as). If not provided, uses the authenticated user's email.",
),
thread_id: Optional[str] = Body(
None, description="Optional Gmail thread ID to reply within."
@@ -905,19 +1010,36 @@ async def send_gmail_message(
references: Optional[str] = Body(
None, description="Optional chain of Message-IDs for proper threading."
),
attachments: Optional[List[Dict[str, str]]] = Body(
None,
description='Optional list of attachments. Each can have: "path" (file path, auto-encodes), OR "content" (standard base64, not urlsafe) + "filename". Optional "mime_type". Example: [{"path": "/path/to/file.pdf"}] or [{"filename": "doc.pdf", "content": "base64data", "mime_type": "application/pdf"}]',
),
) -> str:
"""
Sends an email using the user's Gmail account. Supports both new emails and replies.
Sends an email using the user's Gmail account. Supports both new emails and replies with optional attachments.
Supports Gmail's "Send As" feature to send from configured alias addresses.
Args:
to (str): Recipient email address.
subject (str): Email subject.
body (str): Email body content.
body_format (Literal['plain', 'html']): Email body format. Defaults to 'plain'.
attachments (Optional[List[Dict[str, str]]]): Optional list of attachments. Each dict can contain:
Option 1 - File path (auto-encodes):
- 'path' (required): File path to attach
- 'filename' (optional): Override filename
- 'mime_type' (optional): Override MIME type (auto-detected if not provided)
Option 2 - Base64 content:
- 'content' (required): Standard base64-encoded file content (not urlsafe)
- 'filename' (required): Name of the file
- 'mime_type' (optional): MIME type (defaults to 'application/octet-stream')
cc (Optional[str]): Optional CC email address.
bcc (Optional[str]): Optional BCC email address.
from_name (Optional[str]): Optional sender display name. If provided, the From header will be formatted as 'Name <email>'.
user_google_email (str): The user's Google email address. Required.
from_email (Optional[str]): Optional 'Send As' alias email address. The alias must be
configured in Gmail settings (Settings > Accounts > Send mail as). If not provided,
the email will be sent from the authenticated user's primary email address.
user_google_email (str): The user's Google email address. Required for authentication.
thread_id (Optional[str]): Optional Gmail thread ID to reply within. When provided, sends a reply.
in_reply_to (Optional[str]): Optional Message-ID of the message being replied to. Used for proper threading.
references (Optional[str]): Optional chain of Message-IDs for proper threading. Should include all previous Message-IDs.
@@ -937,6 +1059,14 @@ async def send_gmail_message(
body_format="html"
)
# Send from a configured alias (Send As)
send_gmail_message(
to="user@example.com",
subject="Business Inquiry",
body="Hello from my business address...",
from_email="business@mydomain.com"
)
# Send an email with CC and BCC
send_gmail_message(
to="user@example.com",
@@ -946,6 +1076,28 @@ async def send_gmail_message(
body="Here's the latest update..."
)
# Send an email with attachments (using file path)
send_gmail_message(
to="user@example.com",
subject="Report",
body="Please see attached report.",
attachments=[{
"path": "/path/to/report.pdf"
}]
)
# Send an email with attachments (using base64 content)
send_gmail_message(
to="user@example.com",
subject="Report",
body="Please see attached report.",
attachments=[{
"filename": "report.pdf",
"content": "JVBERi0xLjQK...", # base64 encoded PDF
"mime_type": "application/pdf"
}]
)
# Send a reply
send_gmail_message(
to="user@example.com",
@@ -957,10 +1109,12 @@ async def send_gmail_message(
)
"""
logger.info(
f"[send_gmail_message] Invoked. Email: '{user_google_email}', Subject: '{subject}'"
f"[send_gmail_message] Invoked. Email: '{user_google_email}', Subject: '{subject}', Attachments: {len(attachments) if attachments else 0}"
)
# Prepare the email message
# Use from_email (Send As alias) if provided, otherwise default to authenticated user
sender_email = from_email or user_google_email
raw_message, thread_id_final = _prepare_gmail_message(
subject=subject,
body=body,
@@ -971,8 +1125,9 @@ async def send_gmail_message(
in_reply_to=in_reply_to,
references=references,
body_format=body_format,
from_email=user_google_email,
from_email=sender_email,
from_name=from_name,
attachments=attachments if attachments else None,
)
send_body = {"raw": raw_message}
@@ -986,6 +1141,9 @@ async def send_gmail_message(
service.users().messages().send(userId="me", body=send_body).execute
)
message_id = sent_message.get("id")
if attachments:
return f"Email sent with {len(attachments)} attachment(s)! Message ID: {message_id}"
return f"Email sent! Message ID: {message_id}"
@@ -1005,7 +1163,12 @@ async def draft_gmail_message(
cc: Optional[str] = Body(None, description="Optional CC email address."),
bcc: Optional[str] = Body(None, description="Optional BCC email address."),
from_name: Optional[str] = Body(
None, description="Optional sender display name (e.g., 'Peter Hartree'). If provided, the From header will be formatted as 'Name <email>'."
None,
description="Optional sender display name (e.g., 'Peter Hartree'). If provided, the From header will be formatted as 'Name <email>'.",
),
from_email: Optional[str] = Body(
None,
description="Optional 'Send As' alias email address. Must be configured in Gmail settings (Settings > Accounts > Send mail as). If not provided, uses the authenticated user's email.",
),
thread_id: Optional[str] = Body(
None, description="Optional Gmail thread ID to reply within."
@@ -1016,12 +1179,17 @@ async def draft_gmail_message(
references: Optional[str] = Body(
None, description="Optional chain of Message-IDs for proper threading."
),
attachments: Optional[List[Dict[str, str]]] = Body(
None,
description="Optional list of attachments. Each can have: 'path' (file path, auto-encodes), OR 'content' (standard base64, not urlsafe) + 'filename'. Optional 'mime_type' (auto-detected from path if not provided).",
),
) -> str:
"""
Creates a draft email in the user's Gmail account. Supports both new drafts and reply drafts.
Creates a draft email in the user's Gmail account. Supports both new drafts and reply drafts with optional attachments.
Supports Gmail's "Send As" feature to draft from configured alias addresses.
Args:
user_google_email (str): The user's Google email address. Required.
user_google_email (str): The user's Google email address. Required for authentication.
subject (str): Email subject.
body (str): Email body (plain text).
body_format (Literal['plain', 'html']): Email body format. Defaults to 'plain'.
@@ -1029,9 +1197,21 @@ async def draft_gmail_message(
cc (Optional[str]): Optional CC email address.
bcc (Optional[str]): Optional BCC email address.
from_name (Optional[str]): Optional sender display name. If provided, the From header will be formatted as 'Name <email>'.
from_email (Optional[str]): Optional 'Send As' alias email address. The alias must be
configured in Gmail settings (Settings > Accounts > Send mail as). If not provided,
the draft will be from the authenticated user's primary email address.
thread_id (Optional[str]): Optional Gmail thread ID to reply within. When provided, creates a reply draft.
in_reply_to (Optional[str]): Optional Message-ID of the message being replied to. Used for proper threading.
references (Optional[str]): Optional chain of Message-IDs for proper threading. Should include all previous Message-IDs.
attachments (List[Dict[str, str]]): Optional list of attachments. Each dict can contain:
Option 1 - File path (auto-encodes):
- 'path' (required): File path to attach
- 'filename' (optional): Override filename
- 'mime_type' (optional): Override MIME type (auto-detected if not provided)
Option 2 - Base64 content:
- 'content' (required): Standard base64-encoded file content (not urlsafe)
- 'filename' (required): Name of the file
- 'mime_type' (optional): MIME type (defaults to 'application/octet-stream')
Returns:
str: Confirmation message with the created draft's ID.
@@ -1040,6 +1220,14 @@ async def draft_gmail_message(
# Create a new draft
draft_gmail_message(subject="Hello", body="Hi there!", to="user@example.com")
# Create a draft from a configured alias (Send As)
draft_gmail_message(
subject="Business Inquiry",
body="Hello from my business address...",
to="user@example.com",
from_email="business@mydomain.com"
)
# Create a plaintext draft with CC and BCC
draft_gmail_message(
subject="Project Update",
@@ -1073,7 +1261,7 @@ async def draft_gmail_message(
draft_gmail_message(
subject="Re: Meeting tomorrow",
body="<strong>Thanks for the update!</strong>",
body_format="html,
body_format="html",
to="user@example.com",
thread_id="thread_123",
in_reply_to="<message123@gmail.com>",
@@ -1085,6 +1273,8 @@ async def draft_gmail_message(
)
# Prepare the email message
# Use from_email (Send As alias) if provided, otherwise default to authenticated user
sender_email = from_email or user_google_email
raw_message, thread_id_final = _prepare_gmail_message(
subject=subject,
body=body,
@@ -1095,8 +1285,9 @@ async def draft_gmail_message(
thread_id=thread_id,
in_reply_to=in_reply_to,
references=references,
from_email=user_google_email,
from_email=sender_email,
from_name=from_name,
attachments=attachments,
)
# Create a draft instead of sending
@@ -1111,7 +1302,8 @@ async def draft_gmail_message(
service.users().drafts().create(userId="me", body=draft_body).execute
)
draft_id = created_draft.get("id")
return f"Draft created! Draft ID: {draft_id}"
attachment_info = f" with {len(attachments)} attachment(s)" if attachments else ""
return f"Draft created{attachment_info}! Draft ID: {draft_id}"
def _format_thread_content(thread_data: dict, thread_id: str) -> str: