from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import AsyncIterator
from pydantic import BaseModel
from gmail import Gmail
from models import GetThreadResponse, Message
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
@dataclass
class AppContext:
gmail: Gmail
@asynccontextmanager
async def app_lifespan(_server: FastMCP) -> AsyncIterator[AppContext]:
gmail = Gmail()
yield AppContext(gmail=gmail)
mcp = FastMCP("Gmail Drafter", lifespan=app_lifespan, json_response=True)
class GetUnreadEmailsToolResponse(BaseModel):
sender: str
subject: str
snippet: str
thread_id: str
@staticmethod
def from_message(message: Message) -> "GetUnreadEmailsToolResponse":
return GetUnreadEmailsToolResponse(
sender=message.sender,
subject=message.subject,
snippet=message.snippet,
thread_id=message.thread_id,
)
@mcp.tool()
def get_unread_emails(
ctx: Context[ServerSession, AppContext], max_results: int = 10
) -> list[GetUnreadEmailsToolResponse]:
gmail = ctx.request_context.lifespan_context.gmail
return [
GetUnreadEmailsToolResponse.from_message(message)
for message in gmail.get_unread_messages(max_results)
]
@mcp.tool()
def get_email_thread(
ctx: Context[ServerSession, AppContext], thread_id: str, max_messages: int = 10
) -> GetThreadResponse:
gmail = ctx.request_context.lifespan_context.gmail
return gmail.get_thread(thread_id, max_messages)
@mcp.tool()
async def create_draft_reply(
ctx: Context[ServerSession, AppContext], thread_id: str, body: str
) -> str:
gmail = ctx.request_context.lifespan_context.gmail
gmail.create_draft(thread_id, body)
return "cool"
if __name__ == "__main__":
mcp.run(transport="stdio")
