#!/usr/bin/env python
import os
import base64
import traceback
from typing import Any
from mcp.server.fastmcp import FastMCP
from gmail_client import get_gmail_service, get_docs_service
mcp = FastMCP("gmail-mcp-server")
def google_doc_to_text(doc: dict[str, Any]) -> str:
"""
Convert a Google Docs document resource into plain text.
This walks the document body, concatenating paragraph text.
"""
body = doc.get("body", {})
content = body.get("content", [])
lines: list[str] = []
for element in content:
para = element.get("paragraph")
if not para:
continue
line_parts: list[str] = []
for el in para.get("elements", []):
text_run = el.get("textRun")
if not text_run:
continue
text = text_run.get("content", "")
line_parts.append(text)
# Join runs, strip trailing newlines from Docs
if line_parts:
line = "".join(line_parts).rstrip("\n")
lines.append(line)
return "\n".join(lines).strip()
@mcp.tool()
def get_unread_emails(maxResults: int = 10) -> dict[str, Any]:
"""
Fetch unread emails from the user's inbox.
Returns a JSON structure with:
- id
- threadId
- from
- subject
- snippet
- internalDate
"""
try:
service = get_gmail_service()
results = (
service.users()
.messages()
.list(
userId="me",
labelIds=["INBOX"],
q="is:unread",
maxResults=maxResults,
)
.execute()
)
messages = results.get("messages", [])
emails: list[dict[str, Any]] = []
for msg in messages:
msg_id = msg.get("id")
if not msg_id:
continue
msg_data = (
service.users()
.messages()
.get(userId="me", id=msg_id)
.execute()
)
payload = msg_data.get("payload", {})
headers = payload.get("headers", [])
def get_header(name: str) -> str:
for h in headers:
if h.get("name", "").lower() == name.lower():
return h.get("value", "")
return ""
email = {
"id": msg_data.get("id", ""),
"threadId": msg_data.get("threadId", ""),
"from": get_header("From"),
"subject": get_header("Subject"),
"snippet": msg_data.get("snippet", ""),
"internalDate": msg_data.get("internalDate", ""),
}
emails.append(email)
return {"emails": emails}
except Exception as e:
return {"error": f"Failed to fetch unread emails: {e}"}
@mcp.tool()
def create_draft_reply(threadId: str, replyBody: str) -> dict[str, Any]:
"""
Create a draft reply in Gmail for a given thread.
Inputs:
- threadId: the Gmail thread ID
- replyBody: plain-text body of the reply
Returns:
- draftId
- threadId
- to
- subject
"""
try:
service = get_gmail_service()
thread = (
service.users()
.threads()
.get(userId="me", id=threadId)
.execute()
)
messages = thread.get("messages", [])
if not messages:
return {"error": "Thread not found or empty."}
last_msg = messages[-1]
payload = last_msg.get("payload", {})
headers = payload.get("headers", [])
def get_header(name: str) -> str:
for h in headers:
if h.get("name", "").lower() == name.lower():
return h.get("value", "")
return ""
from_header = get_header("From")
reply_to_header = get_header("Reply-To")
subject_header = get_header("Subject")
message_id_header = get_header("Message-ID")
to = reply_to_header or from_header or ""
clean_subject = subject_header or ""
if clean_subject.lower().startswith("re:"):
clean_subject = clean_subject[3:].strip()
subject = f"Re: {clean_subject}" if clean_subject else "Re:"
mime_lines = [
f"To: {to}" if to else "",
f"Subject: {subject}",
f"In-Reply-To: {message_id_header}" if message_id_header else "",
f"References: {message_id_header}" if message_id_header else "",
'Content-Type: text/plain; charset="UTF-8"',
"",
replyBody,
]
mime_text = "\r\n".join([line for line in mime_lines])
raw_bytes = base64.urlsafe_b64encode(mime_text.encode("utf-8"))
raw_str = raw_bytes.decode("utf-8").rstrip("=")
draft = (
service.users()
.drafts()
.create(
userId="me",
body={
"message": {
"threadId": threadId,
"raw": raw_str,
}
},
)
.execute()
)
return {
"draftId": draft.get("id"),
"threadId": threadId,
"to": to,
"subject": subject,
}
except Exception as e:
return {"error": f"Failed to create draft reply: {e}"}
@mcp.tool()
def get_email_style_guide() -> dict[str, Any]:
"""
Fetch the email style guide from a Google Doc.
Uses the document ID provided in the STYLE_GUIDE_DOC_ID environment variable.
Returns:
{ "styleGuide": "<full plain-text content>" }
"""
try:
doc_id = os.environ.get("STYLE_GUIDE_DOC_ID")
if not doc_id:
return {"error": "STYLE_GUIDE_DOC_ID environment variable is not set."}
docs_service = get_docs_service()
doc = docs_service.documents().get(documentId=doc_id).execute()
text = google_doc_to_text(doc)
return {"styleGuide": text}
except Exception as e:
traceback.print_exc()
return {"error": f"Failed to fetch style guide: {e!r}"}
def main() -> None:
"""Entry point for the MCP server (FastMCP)."""
mcp.run()
if __name__ == "__main__":
main()
