Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/servers/gmail/config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ description: "Interact with Gmail emails and messages"
documentation_path: "README.md"
tools:
- name: "read_emails"
description: "Search and read emails in Gmail with full text body and attachment information"
description: "Search and read emails in Gmail with full text body and attachment information. Supports structured JSON output via output_format parameter."
- name: "send_email"
description: "Send an email through Gmail with optional attachments and delivery tracking"
- name: "forward_email"
Expand Down
128 changes: 119 additions & 9 deletions src/servers/gmail/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -364,7 +364,7 @@ async def handle_list_tools() -> list[Tool]:
return [
Tool(
name="read_emails",
description="Search and read emails in Gmail with full text body and attachment information",
description="Search and read emails in Gmail with full text body and attachment information. Supports structured JSON output for programmatic consumption.",
inputSchema={
"type": "object",
"properties": {
Expand All @@ -374,7 +374,7 @@ async def handle_list_tools() -> list[Tool]:
},
"max_results": {
"type": "integer",
"description": "Maximum number of emails to return (default: 10)",
"description": "Maximum number of emails to return (default: 10, max: 100)",
},
"include_body": {
"type": "boolean",
Expand All @@ -384,6 +384,21 @@ async def handle_list_tools() -> list[Tool]:
"type": "boolean",
"description": "Include attachment information in results (default: true)",
},
"output_format": {
"type": "string",
"enum": ["text", "structured"],
"description": "Output format: 'text' for human-readable (default), 'structured' for machine-readable JSON with typed fields (id, threadId, historyId, from, to, cc, bcc, subject, date, snippet, labels, isUnread, messageId, body, attachments)",
},
"label_ids": {
"type": "array",
"items": {"type": "string"},
"description": "Filter by Gmail label IDs (e.g., ['INBOX', 'UNREAD']). Applied in addition to query. Only used with output_format 'structured'.",
},
"include_headers": {
"type": "array",
"items": {"type": "string"},
"description": "Additional headers to include in structured output (e.g., ['Reply-To', 'In-Reply-To']). From, To, Cc, Bcc, Subject, Date are always included. Only used with output_format 'structured'.",
},
},
"required": ["query"],
},
Expand Down Expand Up @@ -545,18 +560,113 @@ async def handle_call_tool(
raise ValueError("Missing query parameter")

query = arguments["query"]
max_results = int(arguments.get("max_results", 10))
max_results = min(int(arguments.get("max_results", 10)), 100)
include_body = arguments.get("include_body", True)
include_attachments_info = arguments.get("include_attachments_info", True)
output_format = arguments.get("output_format", "text")

results = (
gmail_service.users()
.messages()
.list(userId="me", q=query, maxResults=max_results)
.execute()
)
# Build the list request
list_kwargs = {"userId": "me", "q": query, "maxResults": max_results}
label_ids = arguments.get("label_ids")
if label_ids:
list_kwargs["labelIds"] = label_ids

results = gmail_service.users().messages().list(**list_kwargs).execute()

messages = results.get("messages", [])

# --- Structured JSON output ---
if output_format == "structured":
if not messages:
result_data = {
"emails": [],
"resultCount": 0,
"query": query,
}
return [TextContent(type="text", text=json.dumps(result_data))]

extra_headers = arguments.get("include_headers", [])
standard_headers = [
"From",
"To",
"Cc",
"Bcc",
"Subject",
"Date",
"Message-ID",
]
all_headers = list(
dict.fromkeys(standard_headers + [h for h in extra_headers])
)

email_objects = []
for message in messages:
msg = (
gmail_service.users()
.messages()
.get(userId="me", id=message["id"], format="full")
.execute()
)

headers = {}
for header in msg.get("payload", {}).get("headers", []):
if header["name"] in all_headers:
headers[header["name"]] = header["value"]

labels = msg.get("labelIds", [])

email_obj = {
"id": message["id"],
"threadId": msg.get("threadId", ""),
"historyId": msg.get("historyId", ""),
"from": headers.get("From", ""),
"to": headers.get("To", ""),
"cc": headers.get("Cc", ""),
"bcc": headers.get("Bcc", ""),
"subject": headers.get("Subject", ""),
"date": headers.get("Date", ""),
"snippet": msg.get("snippet", ""),
"labels": labels,
"isUnread": "UNREAD" in labels,
"messageId": headers.get("Message-ID", ""),
}

if extra_headers:
email_obj["extraHeaders"] = {
h: headers.get(h, "") for h in extra_headers
}

if include_body:
payload = msg.get("payload", {})
body = parse_email_body(payload)
email_obj["body"] = {
"text": body.get("text", ""),
"html": body.get("html", ""),
}

if include_attachments_info:
payload = msg.get("payload", {})
attachments = get_attachments_info(payload)
email_obj["attachments"] = [
{
"filename": att["filename"],
"mimeType": att["mimeType"],
"size": att.get("size", 0),
"attachmentId": att.get("attachmentId", ""),
}
for att in attachments
]

email_objects.append(email_obj)

result_data = {
"emails": email_objects,
"resultCount": len(email_objects),
"query": query,
}
return [TextContent(type="text", text=json.dumps(result_data))]

# --- Default text output (unchanged) ---
if not messages:
return [
TextContent(
Expand Down
Loading
Loading